msyavuz commented on code in PR #42305:
URL: https://github.com/apache/superset/pull/42305#discussion_r3665262222
##########
superset-frontend/src/middleware/asyncEvent.ts:
##########
@@ -161,11 +171,13 @@ export const waitForAsyncData = async (
}
};
- // When the caller aborts (chart superseded/unmounted), stop listening so
the
- // listener and its retained closure don't leak and keep the poller busy.
+ // When the caller aborts (Stop pressed, chart superseded/unmounted), stop
+ // listening so the listener and its retained closure don't leak, and ask
the
+ // server to cancel the job so it stops consuming warehouse resources.
if (signal) {
onAbort = () => {
cleanup();
+ cancelAsyncJob(jobId);
reject(new DOMException('Aborted', 'AbortError'));
};
Review Comment:
`cleanup()` runs after the await, so an abort mid-fetch still rejects with
an AbortError and cancels the job.
##########
superset/async_events/async_query_manager.py:
##########
@@ -385,3 +417,79 @@ def update_job(
self._cache.xadd(scoped_stream_name, event_data, "*",
self._stream_limit)
self._cache.xadd(full_stream_name, event_data, "*",
self._stream_limit_firehose)
+
+ # Once a job reaches a terminal state its cancel record is dead weight
+ # (and a stale record would let a caller "cancel" a finished job), so
+ # drop it. TTL is the backstop if this is never reached.
+ if status in (self.STATUS_DONE, self.STATUS_ERROR,
self.STATUS_CANCELLED):
+ if job_id := job_metadata.get("job_id"):
+ self._cache.delete(self._job_registry_key(job_id))
+
+ def is_job_cancelled(self, job_id: str) -> bool:
+ """
+ Whether ``cancel_job`` has flagged this job for cancellation.
+
+ Called from the worker's exception handler, so any cache failure is
+ swallowed and treated as "not cancelled" — a Redis blip must never mask
+ the original error (e.g. a genuine timeout) with a connection error.
+ """
+ if not self._cache:
+ return False
+ try:
+ raw = self._cache.get(self._job_registry_key(job_id))
+ if raw is None:
+ return False
+ return bool(json.loads(raw).get("cancelled"))
+ except Exception: # pylint: disable=broad-except
+ logger.warning(
+ "Failed to read cancellation flag for job %s", job_id,
exc_info=True
+ )
+ return False
+
+ def cancel_job(self, job_id: str, channel_id: str, user_id: Optional[int])
-> None:
+ """
+ Authorize and cancel a running async job.
+
+ The caller's ``channel_id`` and ``user_id`` (resolved server-side from
+ the request, never taken from the client) must match the job's original
+ owner. On success the running Celery task is revoked; the SIGUSR1 it
+ receives surfaces as ``SoftTimeLimitExceeded`` in the worker, which
+ reads the cancelled flag set here and emits the terminal
+ ``STATUS_CANCELLED`` event — so exactly one terminal event is written.
+
+ :raises AsyncQueryJobException: the job is unknown or already terminal
+ :raises AsyncQueryTokenException: the caller does not own the job
+ """
+ if not self._cache:
+ raise CacheBackendNotInitialized("Cache backend not initialized")
+
+ key = self._job_registry_key(job_id)
+ raw = self._cache.get(key)
+ if raw is None:
+ raise AsyncQueryJobException("Job not found or already completed")
+
+ record = json.loads(raw)
+ if record.get("channel_id") != channel_id or record.get("user_id") !=
user_id:
+ raise AsyncQueryTokenException("Not authorized to cancel this job")
+
+ # Flag before revoking so the worker's timeout handler, which may fire
+ # almost immediately, reliably sees the cancellation. Write only if the
+ # key still exists (``xx``): if the job finished and cleared its record
+ # between the read above and here, don't recreate a stale record or
+ # revoke a task that is already gone — report it as not found instead.
+ flagged = self._cache.set(
+ key,
+ json.dumps({**record, "cancelled": True}),
+ ex=self._jwt_expiration_seconds or None,
+ xx=True,
+ )
+ if not flagged:
+ raise AsyncQueryJobException("Job not found or already completed")
Review Comment:
The record is deleted before the result event now, so a cancel racing a
completion fails its conditional write and 404s.
##########
superset/async_events/async_query_manager.py:
##########
@@ -385,3 +417,79 @@ def update_job(
self._cache.xadd(scoped_stream_name, event_data, "*",
self._stream_limit)
self._cache.xadd(full_stream_name, event_data, "*",
self._stream_limit_firehose)
+
+ # Once a job reaches a terminal state its cancel record is dead weight
+ # (and a stale record would let a caller "cancel" a finished job), so
+ # drop it. TTL is the backstop if this is never reached.
+ if status in (self.STATUS_DONE, self.STATUS_ERROR,
self.STATUS_CANCELLED):
+ if job_id := job_metadata.get("job_id"):
+ self._cache.delete(self._job_registry_key(job_id))
+
+ def is_job_cancelled(self, job_id: str) -> bool:
+ """
+ Whether ``cancel_job`` has flagged this job for cancellation.
+
+ Called from the worker's exception handler, so any cache failure is
+ swallowed and treated as "not cancelled" — a Redis blip must never mask
+ the original error (e.g. a genuine timeout) with a connection error.
+ """
+ if not self._cache:
+ return False
+ try:
+ raw = self._cache.get(self._job_registry_key(job_id))
+ if raw is None:
+ return False
+ return bool(json.loads(raw).get("cancelled"))
+ except Exception: # pylint: disable=broad-except
+ logger.warning(
+ "Failed to read cancellation flag for job %s", job_id,
exc_info=True
+ )
+ return False
+
+ def cancel_job(self, job_id: str, channel_id: str, user_id: Optional[int])
-> None:
+ """
+ Authorize and cancel a running async job.
+
+ The caller's ``channel_id`` and ``user_id`` (resolved server-side from
+ the request, never taken from the client) must match the job's original
+ owner. On success the running Celery task is revoked; the SIGUSR1 it
+ receives surfaces as ``SoftTimeLimitExceeded`` in the worker, which
+ reads the cancelled flag set here and emits the terminal
+ ``STATUS_CANCELLED`` event — so exactly one terminal event is written.
+
+ :raises AsyncQueryJobException: the job is unknown or already terminal
+ :raises AsyncQueryTokenException: the caller does not own the job
+ """
+ if not self._cache:
+ raise CacheBackendNotInitialized("Cache backend not initialized")
+
+ key = self._job_registry_key(job_id)
+ raw = self._cache.get(key)
+ if raw is None:
+ raise AsyncQueryJobException("Job not found or already completed")
+
+ record = json.loads(raw)
+ if record.get("channel_id") != channel_id or record.get("user_id") !=
user_id:
+ raise AsyncQueryTokenException("Not authorized to cancel this job")
+
+ # Flag before revoking so the worker's timeout handler, which may fire
+ # almost immediately, reliably sees the cancellation. Write only if the
+ # key still exists (``xx``): if the job finished and cleared its record
+ # between the read above and here, don't recreate a stale record or
+ # revoke a task that is already gone — report it as not found instead.
+ flagged = self._cache.set(
+ key,
+ json.dumps({**record, "cancelled": True}),
+ ex=self._jwt_expiration_seconds or None,
+ xx=True,
+ )
+ if not flagged:
+ raise AsyncQueryJobException("Job not found or already completed")
+
+ # pylint: disable=import-outside-toplevel
+ from superset.extensions import celery_app
+
+ # SIGUSR1 raises SoftTimeLimitExceeded inside the running task rather
+ # than hard-killing the process, so the worker can emit its terminal
+ # event before exiting.
+ celery_app.control.revoke(job_id, terminate=True, signal="SIGUSR1")
Review Comment:
Real gap, fixed: the cancel request emits the terminal event itself, since a
task revoked while queued never reaches the worker handler.
##########
superset/async_events/api.py:
##########
@@ -99,3 +104,76 @@ def events(self) -> Response:
return self.response_401()
return self.response(200, result=events)
+
+ @expose("/<job_id>/cancel", methods=("POST",))
+ @event_logger.log_this
+ @protect()
+ @safe
+ @statsd_metrics
+ @permission_name("list")
Review Comment:
Switched to a dedicated `can_cancel` permission, which needs `superset init`
to be granted.
##########
superset/tasks/async_queries.py:
##########
@@ -84,6 +84,23 @@ def _load_user_from_job_metadata(job_metadata: dict[str,
Any]) -> User:
return user
+def _handle_soft_time_limit(
+ job_metadata: dict[str, Any], ex: Exception, activity: str
+) -> None:
+ """
+ SoftTimeLimitExceeded is raised both by a genuine timeout and by a
+ user-initiated cancel (revoke sends SIGUSR1). Emit the matching terminal
+ event so a cancelled job is reported as cancelled.
+ """
+ if async_query_manager.is_job_cancelled(job_metadata["job_id"]):
+ logger.info("Cancelled by the user while %s", activity)
+ async_query_manager.update_job(
+ job_metadata, async_query_manager.STATUS_CANCELLED
+ )
+ else:
Review Comment:
Yes, a genuine timeout now emits STATUS_ERROR, and cancelled jobs are
reported by the cancel request instead.
##########
superset-frontend/src/middleware/asyncEvent.ts:
##########
@@ -104,6 +104,15 @@ const fetchCachedData = async (
return { status, data };
};
+const cancelAsyncJob = (jobId: string) => {
+ // Best-effort server-side cancel; the request stops the running Celery task
+ // so it no longer consumes warehouse resources. Failures are non-fatal: the
+ // client has already stopped waiting on the job.
+ SupersetClient.post({
+ endpoint: `/api/v1/async_event/${jobId}/cancel`,
+ }).catch(() => {});
Review Comment:
Added a `logging.warn` in the catch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
