You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by en...@apache.org on 2026/09/21 15:44:15 UTC
(superset) branch master updated: fix(reports): enable alert retries with execution safeguards (#44336)
This is an automated email from the ASF dual-hosted git repository.
EnxDev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new 8ec8beecffd fix(reports): enable alert retries with execution safeguards (#44336)
8ec8beecffd is described below
commit 8ec8beecffdf91b1efba4fa121330a2517c355c5
Author: Mafi <ma...@gmail.com>
AuthorDate: Tue Sep 22 01:44:07 2026 +1000
fix(reports): enable alert retries with execution safeguards (#44336)
Co-authored-by: Enzo Martellucci <52...@users.noreply.github.com>
---
UPDATING.md | 40 ++
docs/admin_docs/configuration/alerts-reports.mdx | 32 +-
.../src/features/alerts/AlertReportModal.test.tsx | 57 +-
.../src/features/alerts/AlertReportModal.tsx | 6 +-
superset/commands/report/execute.py | 415 +++++++++--
superset/commands/report/execution_claim.py | 199 ++++++
superset/commands/report/update.py | 6 -
..._93d1b4a76c02_add_report_execution_ownership.py | 55 ++
superset/reports/models.py | 2 +
superset/reports/notifications/base.py | 1 +
superset/reports/notifications/email.py | 16 +-
superset/reports/notifications/slack_mixin.py | 6 +-
superset/reports/notifications/webhook.py | 9 +-
superset/reports/schemas.py | 7 -
superset/tasks/scheduler.py | 13 +
superset/utils/report_execution.py | 15 +
tests/integration_tests/reports/commands_tests.py | 104 ++-
.../reports/execution_claim_tests.py | 32 +
tests/unit_tests/commands/report/execute_test.py | 788 ++++++++++++++++++++-
.../commands/report/execution_claim_test.py | 412 +++++++++++
.../report/execution_ownership_migration_test.py | 87 +++
tests/unit_tests/commands/report/update_test.py | 10 +-
.../reports/notifications/email_tests.py | 38 +-
.../reports/notifications/slack_tests.py | 18 +
.../reports/notifications/webhook_tests.py | 71 +-
tests/unit_tests/reports/schemas_test.py | 28 +
.../tasks/test_scheduler_soft_timeout.py | 17 +
27 files changed, 2305 insertions(+), 179 deletions(-)
diff --git a/UPDATING.md b/UPDATING.md
index 81b56b62ef5..cf3d4bd6bab 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -24,6 +24,46 @@ assists people when migrating to a new version.
## Next
+### Scheduled report and alert retry admission
+
+Run `superset db upgrade` before starting workers with this version. The migration
+adds nullable `execution_owner` and `execution_window` columns to `report_schedule`.
+Pause scheduling and drain in-flight executions and queued retry tasks before
+upgrading workers together: older workers do not participate in execution fencing.
+Two-argument retries without ownership evidence are discarded. Older one-argument
+retry messages cannot be distinguished from fresh cron tasks and must be drained;
+they are not reliably discarded by the new workers. Restart scheduling after
+migration and worker replacement.
+Mixed-version workers are not supported for this transition: old workers cannot
+consume the new retry task signature. Drain queued retries as well as active jobs
+before replacement. New workers log `report_retry_discarded` with
+`reason=missing_execution_owner` and increment
+`reports.execute.legacy_retry_discarded` for legacy retry tasks. Rerun any affected
+schedule after the upgrade; discarded retries are not automatically replayed.
+
+In the other direction, a new worker can enqueue a three-argument retry that an
+old worker cannot consume: the old worker raises `TypeError`, loses that retry,
+and can leave the schedule in `RETRYING` until stale-retry recovery. Do not perform
+a mixed-version rolling deployment; stop scheduling, drain active jobs and queued
+retries, then replace all consumers before restarting scheduling.
+
+With `ALERT_REPORTS_RETRY` enabled, alerts can opt into retries as reports do.
+Retries re-evaluate alert conditions. Whole-execution retries stop once delivery
+has started because a failed send may already have reached a recipient. Retry
+notifications no longer include raw provider diagnostics; consult execution logs.
+Ordinary editor-only error emails retain HTML-sanitized diagnostics. Notifications
+to configured recipients, including retry/final-failure notices, remain redacted.
+
+Ownership is rechecked in a short committed transaction before each recipient.
+No schedule-row lock is held during SMTP/Slack I/O. Recovery can fence subsequent
+sends, but cannot recall a notification already in flight; delivery is not atomic
+with database ownership and exactly-once delivery is not guaranteed.
+
+If the fallback ERROR-log transaction cannot commit, the execution still depends
+on working-timeout recovery. Error-notification delivery and its database marker
+are not atomic: a send followed by a marker-write failure can result in a duplicate
+error notice. This does not authorize replay of data-bearing notifications.
+
- With `SEMANTIC_LAYERS` enabled, combined connection discovery honors `Database.can_read` and `SemanticLayer.can_read` independently. Each permitted source retains its normal row filters, including dynamic database filters for Admin. A source filter never includes rows or counts from a denied source; callers with neither read permission are denied. Feature-off database browsing is unchanged.
- The combined datasource list (`GET /api/v1/datasource/`) accepts Dataset read without an additional Datasource read grant, regardless of `SEMANTIC_LAYERS`. With the flag enabled, SemanticView read independently permits semantic-view discovery. Existing row-level dataset/chart access remains enforced.
- The `presto` extra requires PyHive 0.7.0 or later. PyHive 0.6.5 cannot load
diff --git a/docs/admin_docs/configuration/alerts-reports.mdx b/docs/admin_docs/configuration/alerts-reports.mdx
index 76f1196a80a..93b0835e3df 100644
--- a/docs/admin_docs/configuration/alerts-reports.mdx
+++ b/docs/admin_docs/configuration/alerts-reports.mdx
@@ -505,7 +505,37 @@ captured during the stability window. Successful checks emit
`report_capture_readiness_stable`; a timeout emits
`report_capture_readiness_changed` and stops capture and delivery.
-### Confirm feature flag is enabled and you have sufficient permissions
+### Whole-execution retries
+
+Enable `ALERT_REPORTS_RETRY` and select **Enable Retries** in each alert or report's
+error-handling settings. This does not opt existing schedules in automatically.
+The default allows three additional attempts, with exponential delays starting at
+60 seconds and capped by `ALERT_REPORTS_RETRY_MAX_DELAY_SECONDS`. Screenshot-tile
+retries are separate from these whole-execution attempts.
+
+Each retry gets a fresh browser context and re-evaluates an alert's condition; if
+the condition no longer fires, it finishes without sending the alert. Duplicate
+or obsolete retry messages cannot claim the same execution window. The worker
+logs `report_retry_discarded` or `report_execution_not_claimed` when admission fails.
+
+Whole-execution retry applies before notification delivery starts. A delivery
+failure is not replayed, since a provider may have accepted the notification
+before returning an error. It is not an exactly-once delivery guarantee, nor a
+guarantee of recovery after a worker hard kill. Optional retry/final-failure
+notifications are text-only and omit raw provider diagnostics. Rejected PDF/PNG
+attachments are never attached to those notifications.
+
+Ordinary error emails sent only to schedule editors retain HTML-sanitized diagnostic
+text. Retry and final-failure notices sent to configured recipients omit it.
+Turning off retries cancels a matching queued retry without blocking the next cron
+window. Abandoned retry chains become eligible for recovery after
+`ALERT_REPORTS_RETRY_MAX_DELAY_SECONDS`. Alerts retain their per-schedule Celery
+timeout rather than inheriting the global report execution budget.
+Rendered alert attachments have a separate browser-capture budget, capped by the
+remaining alert time and the configured report execution budget. Time spent
+evaluating the alert condition does not consume that separate capture budget.
+
+### Feature flag and permissions checks
If you don't see "Alerts & Reports" under the _Manage_ section of the Settings dropdown in the Superset UI, you need to enable the `ALERT_REPORTS` feature flag (see above). Enable another feature flag and check to see that it took effect, to verify that your config file is getting loaded.
diff --git a/superset-frontend/src/features/alerts/AlertReportModal.test.tsx b/superset-frontend/src/features/alerts/AlertReportModal.test.tsx
index a5a4923072c..8e3136dc804 100644
--- a/superset-frontend/src/features/alerts/AlertReportModal.test.tsx
+++ b/superset-frontend/src/features/alerts/AlertReportModal.test.tsx
@@ -498,13 +498,13 @@ test('renders 5 sections for reports', () => {
expect(sections.length).toBe(5);
});
-test('renders 5 sections for alerts', () => {
+test('renders 6 sections for alerts', () => {
render(<AlertReportModal {...generateMockedProps(false)} />, {
useRedux: true,
});
const sections = screen.getAllByRole('tab');
- expect(sections.length).toBe(5);
+ expect(sections.length).toBe(6);
});
// Validation
@@ -1579,12 +1579,12 @@ test('adding and removing dashboard filter rows', async () => {
});
test('alert shows condition section, report does not', () => {
- // Alert has 5 sections (general, condition, content, schedule, notification)
+ // Alerts also include condition and error-handling sections.
const { unmount } = render(
<AlertReportModal {...generateMockedProps(false)} />,
{ useRedux: true },
);
- expect(screen.getAllByRole('tab')).toHaveLength(5);
+ expect(screen.getAllByRole('tab')).toHaveLength(6);
expect(screen.getByTestId('alert-condition-panel')).toBeInTheDocument();
unmount();
@@ -2985,7 +2985,7 @@ test('clearing notification recipients disables submit and prevents API call', a
useRedux: true,
});
- // Wait for all validation to pass (5 checkmarks = fully valid alert)
+ // Wait for all alert sections to validate.
await waitFor(() => {
expect(
screen.queryAllByRole('img', { name: /check-circle/i }),
@@ -3048,30 +3048,35 @@ test('renders error handling panel with Enable Retries switch', async () => {
expect(screen.getByText('Enable Retries')).toBeInTheDocument();
});
-test('shows retry options when Enable Retries is toggled on', async () => {
- render(<AlertReportModal {...generateMockedProps(true)} />, {
- useRedux: true,
- });
- const errorHandlingTab = screen.getByText('Error handling');
- await userEvent.click(errorHandlingTab);
+test.each([false, true])(
+ 'shows retry options for isReport=%s',
+ async isReport => {
+ render(<AlertReportModal {...generateMockedProps(isReport)} />, {
+ useRedux: true,
+ });
+ const errorHandlingTab = screen.getByText('Error handling');
+ await userEvent.click(errorHandlingTab);
- // Retry options should not be visible initially
- expect(screen.queryByText('Maximum Retry Attempts')).not.toBeInTheDocument();
+ // Retry options should not be visible initially
+ expect(
+ screen.queryByText('Maximum Retry Attempts'),
+ ).not.toBeInTheDocument();
- // Toggle Enable Retries — the Switch renders as a <button role="switch">
- const switches = screen.getAllByRole('switch');
- const enableRetriesSwitch = switches[switches.length - 1];
- await userEvent.click(enableRetriesSwitch);
+ // Toggle Enable Retries — the Switch renders as a <button role="switch">
+ const switches = screen.getAllByRole('switch');
+ const enableRetriesSwitch = switches[switches.length - 1];
+ await userEvent.click(enableRetriesSwitch);
- // Retry options should now be visible
- await waitFor(() => {
- expect(screen.getByText('Maximum Retry Attempts')).toBeInTheDocument();
- expect(screen.getByText('Send Failed Reports')).toBeInTheDocument();
- expect(screen.getByText('Failure Notifications')).toBeInTheDocument();
- expect(screen.getByText('Owners')).toBeInTheDocument();
- expect(screen.getByText('Report Recipients')).toBeInTheDocument();
- });
-});
+ // Retry options should now be visible
+ await waitFor(() => {
+ expect(screen.getByText('Maximum Retry Attempts')).toBeInTheDocument();
+ expect(screen.getByText('Send Failed Reports')).toBeInTheDocument();
+ expect(screen.getByText('Failure Notifications')).toBeInTheDocument();
+ expect(screen.getByText('Owners')).toBeInTheDocument();
+ expect(screen.getByText('Report Recipients')).toBeInTheDocument();
+ });
+ },
+);
test('hides retry options and resets state when Enable Retries is toggled off', async () => {
render(<AlertReportModal {...generateMockedProps(true)} />, {
diff --git a/superset-frontend/src/features/alerts/AlertReportModal.tsx b/superset-frontend/src/features/alerts/AlertReportModal.tsx
index ba923d081b9..daaca918549 100644
--- a/superset-frontend/src/features/alerts/AlertReportModal.tsx
+++ b/superset-frontend/src/features/alerts/AlertReportModal.tsx
@@ -2764,7 +2764,7 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
</>
),
},
- ...(isReport && isFeatureEnabled(FeatureFlag.AlertReportsRetry)
+ ...(isFeatureEnabled(FeatureFlag.AlertReportsRetry)
? [
{
key: 'error-handling',
@@ -2772,7 +2772,7 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
<CollapseLabelInModal
title={t('Error handling')}
subtitle={t(
- 'Configure retry behavior on delivery failure.',
+ 'Configure retries when alert or report generation fails before delivery.',
)}
testId="error-handling-panel"
/>
@@ -2804,7 +2804,7 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
</div>
<InfoTooltip
tooltip={t(
- 'Automatically retry sending the report when delivery fails.',
+ 'Retry generation failures before delivery starts. Alerts re-check their condition on each attempt. Delivery failures are not replayed.',
)}
/>
</StyledSwitchContainer>
diff --git a/superset/commands/report/execute.py b/superset/commands/report/execute.py
index 9c0f77d0d97..75177eec87e 100644
--- a/superset/commands/report/execute.py
+++ b/superset/commands/report/execute.py
@@ -20,6 +20,7 @@ import urllib.parse
import urllib.request
from collections.abc import Sequence
from contextlib import closing
+from dataclasses import replace
from datetime import datetime, timedelta, timezone
from typing import Any, Optional, TYPE_CHECKING, Union
from urllib.error import URLError
@@ -61,6 +62,12 @@ from superset.commands.report.exceptions import (
ReportScheduleXlsxFailedError,
ReportScheduleXlsxTimeout,
)
+from superset.commands.report.execution_claim import (
+ cancel_disabled_retry,
+ claim_execution,
+ ExecutionClaim,
+ normalize_window,
+)
from superset.commands.report.slack_upgrade import SlackV1UpgradeCoordinator
from superset.common.chart_data import ChartDataResultFormat, ChartDataResultType
from superset.daos.report import (
@@ -103,6 +110,7 @@ from superset.utils.decorators import (
from superset.utils.file import sanitize_title
from superset.utils.pdf import build_pdf_from_screenshots
from superset.utils.report_execution import (
+ get_report_task_timeout_options,
ReportExecutionBudgetExceededError,
ReportExecutionContext,
ReportExecutionDeadline,
@@ -142,26 +150,49 @@ def resolve_executor_user(model: ReportSchedule) -> tuple["User", str]:
def _should_build_execution_context(model: ReportSchedule) -> bool:
- """
- Whether an execution should run under a :class:`ReportExecutionContext`.
-
- Reports always do — their behavior is unchanged. Alerts join them only when
- they deliver a rendered PNG/PDF screenshot to recipients, which happens when
- ``ALERTS_ATTACH_REPORTS`` is enabled. Delivered screenshots must fail closed:
- the context selects the fail-closed readiness predicate and disables
- partial-tile fallback, so a blank or incomplete capture raises instead of
- being delivered.
-
- CSV/text alerts, alerts without the attach flag, the non-delivered
- query-context capture, and UI thumbnails are deliberately excluded and keep
- their lenient capture contract.
- """
+ """Give every scheduled report and alert a shared deadline and ownership context."""
+ return model.type in (ReportScheduleType.REPORT, ReportScheduleType.ALERT)
+
+
+def _uses_report_capture_contract(model: ReportSchedule) -> bool:
+ """Keep ownership separate from the existing rendered-alert capture policy."""
+ return model.type == ReportScheduleType.REPORT or (
+ model.report_format in (ReportDataFormat.PNG, ReportDataFormat.PDF)
+ and feature_flag_manager.is_feature_enabled("ALERTS_ATTACH_REPORTS")
+ )
+
+
+def _execution_budget_seconds(model: ReportSchedule) -> float:
+ """Preserve alert Celery limits rather than applying the global report budget."""
if model.type == ReportScheduleType.REPORT:
- return True
- return model.report_format in (
- ReportDataFormat.PNG,
- ReportDataFormat.PDF,
- ) and feature_flag_manager.is_feature_enabled("ALERTS_ATTACH_REPORTS")
+ return resolve_report_execution_budget_seconds(
+ app.config, working_timeout=model.working_timeout
+ )
+ return float(
+ get_report_task_timeout_options(
+ is_report=False, working_timeout=model.working_timeout, config=app.config
+ ).get("soft_time_limit", float("inf"))
+ )
+
+
+def _capture_execution_context(
+ model: ReportSchedule, context: ReportExecutionContext | None
+) -> ReportExecutionContext | None:
+ """Keep data-only alerts lenient and bound rendered alerts' browser phase."""
+ if not _uses_report_capture_contract(model):
+ return None
+ if context and model.type == ReportScheduleType.ALERT:
+ # An unlimited alert must never pass infinity to Playwright. Browser
+ # work has its own budget without counting time spent on the alert query.
+ return context.with_deadline(
+ ReportExecutionDeadline(
+ total_seconds=min(
+ context.deadline.available_seconds("screenshot_capture"),
+ resolve_report_execution_budget_seconds(app.config),
+ )
+ )
+ )
+ return context
def log_report_delivery_phase(
@@ -215,6 +246,68 @@ def persist_owned_report_execution_terminal_error(
# this boundary. Roll back again so a failed terminal flush cannot leave
# the scoped session unusable for the retry.
db.session.rollback() # pylint: disable=consider-using-transaction
+ if report_context is not None and report_context.execution_claimed:
+ schedule = (
+ db.session.query(ReportSchedule)
+ .filter_by(id=report_schedule_id)
+ .one_or_none()
+ )
+ if schedule is None:
+ return False
+ matched = (
+ db.session.query(ReportSchedule)
+ .filter(
+ ReportSchedule.id == report_schedule_id,
+ ReportSchedule.execution_owner == str(execution_id),
+ ReportSchedule.last_state == ReportState.WORKING,
+ )
+ .update(
+ {
+ ReportSchedule.last_state: ReportState.ERROR,
+ ReportSchedule.last_eval_dttm: datetime.now(
+ timezone.utc
+ ).replace(tzinfo=None),
+ },
+ synchronize_session="fetch",
+ )
+ )
+ if matched != 1:
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ return False
+ # The claim is durable even if the WORKING log was rolled back.
+ # create_log promotes that log when present and creates it otherwise.
+ state = BaseReportState(
+ schedule,
+ schedule.execution_window,
+ execution_id,
+ report_context,
+ )
+ state.create_log(error_message, log_state=ReportState.ERROR)
+ logger.error(
+ "report_execution_terminal %s state=Error terminal_reason=%s",
+ report_context.log_context,
+ terminal_reason,
+ )
+ # Persist first; notification failure must not undo terminalization.
+ # Timeout cleanup must not start another network operation.
+ if terminal_reason not in (
+ SoftTimeLimitExceeded.__name__,
+ ReportExecutionBudgetExceededError.__name__,
+ ):
+ try:
+ if not state.is_in_error_grace_period():
+ state.send_error(
+ f"Error occurred for {schedule.type}: {schedule.name}",
+ error_message,
+ )
+ state.create_log(
+ REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER,
+ log_state=ReportState.ERROR,
+ )
+ except Exception: # noqa: BLE001
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ logger.exception("Failed fallback error notification")
+ return True
working_log = (
db.session.query(ReportExecutionLog)
.filter(
@@ -243,14 +336,36 @@ def persist_owned_report_execution_terminal_error(
report_schedule.last_state == ReportState.WORKING
and latest_working_log is not None
and latest_working_log.uuid == execution_id
+ and (
+ report_context is None
+ or not report_context.execution_claimed
+ or report_schedule.execution_owner == str(execution_id)
+ )
)
ended_at = datetime.now(timezone.utc).replace(tzinfo=None)
working_log.state = ReportState.ERROR
working_log.error_message = error_message
working_log.end_dttm = ended_at
if owns_schedule_state:
- report_schedule.last_state = ReportState.ERROR
- report_schedule.last_eval_dttm = ended_at
+ with db.session.no_autoflush:
+ owns_schedule_state = (
+ db.session.query(ReportSchedule)
+ .filter(
+ ReportSchedule.id == report_schedule_id,
+ ReportSchedule.last_state == ReportState.WORKING,
+ ReportSchedule.execution_owner
+ == report_schedule.execution_owner,
+ ReportSchedule.last_eval_dttm == report_schedule.last_eval_dttm,
+ )
+ .update(
+ {
+ ReportSchedule.last_state: ReportState.ERROR,
+ ReportSchedule.last_eval_dttm: ended_at,
+ },
+ synchronize_session=False,
+ )
+ == 1
+ )
db.session.commit() # pylint: disable=consider-using-transaction
log_context = (
@@ -312,6 +427,7 @@ class BaseReportState:
self._start_dttm: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
self._execution_id = execution_id
self._report_execution_context = report_execution_context
+ self._delivery_started = False
self._execution_warnings: list[str] = []
self._slack_v1_upgrade = SlackV1UpgradeCoordinator(
report_schedule,
@@ -363,11 +479,13 @@ class BaseReportState:
) -> float | None:
if not self._report_execution_context:
return requested_seconds
- return self._report_execution_context.deadline.timeout_seconds(
+ timeout = self._report_execution_context.deadline.timeout_seconds(
phase,
requested_seconds=requested_seconds,
reserve_seconds=reserve_seconds,
)
+ # Socket transports represent an unbounded wait as None, not infinity.
+ return None if timeout == float("inf") else timeout
def update_report_schedule_and_log(
self,
@@ -406,6 +524,29 @@ class BaseReportState:
execution log.
"""
+ context = self._report_execution_context
+ if context is not None and context.execution_claimed:
+ with db.session.no_autoflush:
+ matched = (
+ db.session.query(ReportSchedule)
+ .filter(
+ ReportSchedule.id == self._report_schedule.id,
+ ReportSchedule.execution_owner == str(self._execution_id),
+ )
+ .update(
+ {
+ ReportSchedule.last_state: state,
+ ReportSchedule.last_eval_dttm: datetime.now(
+ timezone.utc
+ ).replace(tzinfo=None),
+ },
+ synchronize_session=False,
+ )
+ )
+ if matched != 1:
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ raise ReportSchedulePreviousWorkingError()
+
if state == ReportState.WORKING:
self._report_schedule.last_value = None
self._report_schedule.last_value_row_json = None
@@ -759,6 +900,32 @@ class BaseReportState:
for tab_anchor in tab_anchors
]
+ def _assert_execution_owned(self) -> None:
+ """Do not deliver after recovery or a successor has fenced this worker."""
+ context = self._report_execution_context
+ if context is None or not context.execution_claimed:
+ return
+ # This no-op UPDATE needs matched-row counts, not changed-row counts.
+ # SQLAlchemy's MySQL dialect enables CLIENT_FOUND_ROWS for this contract.
+ with db.session.no_autoflush:
+ owned = (
+ db.session.query(ReportSchedule)
+ .filter(
+ ReportSchedule.id == self._report_schedule.id,
+ ReportSchedule.execution_owner == str(self._execution_id),
+ )
+ .update(
+ {ReportSchedule.execution_owner: str(self._execution_id)},
+ synchronize_session=False,
+ )
+ )
+ if owned != 1:
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ raise ReportSchedulePreviousWorkingError()
+ # A CAS UPDATE also locks the row. Release it before network I/O,
+ # including any pending state changes protected by this ownership check.
+ db.session.commit() # pylint: disable=consider-using-transaction
+
def _get_screenshots(self) -> list[bytes]:
"""
Get chart or dashboard screenshots
@@ -768,6 +935,10 @@ class BaseReportState:
user, _ = resolve_executor_user(self._report_schedule)
+ capture_context = _capture_execution_context(
+ self._report_schedule, self._report_execution_context
+ )
+
max_width = app.config["ALERT_REPORTS_MAX_CUSTOM_SCREENSHOT_WIDTH"]
if self._report_schedule.chart:
@@ -808,7 +979,7 @@ class BaseReportState:
imge = screenshot.get_screenshot(
user=user,
log_context=self._log_context,
- report_execution_context=self._report_execution_context,
+ report_execution_context=capture_context,
)
if imge is None:
raise ReportScheduleScreenshotFailedError(
@@ -1423,6 +1594,7 @@ class BaseReportState:
report_context = getattr(self, "_report_execution_context", None)
self._assert_rendered_delivery_allowed(notification_content)
for recipient in recipients:
+ self._assert_execution_owned()
try:
log_report_delivery_phase(
report_context,
@@ -1430,6 +1602,9 @@ class BaseReportState:
"start",
enforce_budget=True,
)
+ # Delivery outcomes may be ambiguous; do not replay the whole
+ # execution after a notification may have been accepted.
+ self._delivery_started = True
self._send_notification(notification_content, recipient)
log_report_delivery_phase(
report_context,
@@ -1498,6 +1673,7 @@ class BaseReportState:
notification_content = NotificationContent(
name=sanitize_title(name),
text=message,
+ is_editor_error=True,
header_data=header_data,
url=url,
# NULL (rows predating the include_cta column) is treated as True
@@ -1558,9 +1734,7 @@ class BaseReportState:
be compared safely. MySQL DateTime columns truncate microseconds,
so without this the round-tripped anchor would differ from the
in-memory value."""
- if dt is not None:
- return dt.replace(tzinfo=None, microsecond=0)
- return dt
+ return normalize_window(dt)
def _is_retry_window_stale(self) -> bool:
"""
@@ -1599,8 +1773,17 @@ class BaseReportState:
# shares the same window identity. This lets _is_retry_window_stale()
# detect when a *new* crontab window fires while retries are in-flight.
execute_task.apply_async(
- (self._report_schedule.id, self._scheduled_dttm.isoformat()),
+ (
+ self._report_schedule.id,
+ self._scheduled_dttm.isoformat(),
+ str(self._execution_id),
+ ),
countdown=delay_seconds,
+ **get_report_task_timeout_options(
+ is_report=self._report_schedule.type == ReportScheduleType.REPORT,
+ working_timeout=self._report_schedule.working_timeout,
+ config=app.config,
+ ),
)
def send_retry_notification(
@@ -1634,7 +1817,7 @@ class BaseReportState:
url = self._get_url(user_friendly=True)
notification_content = NotificationContent(
name=sanitize_title(self._report_schedule.name),
- text=error_message,
+ text="Report execution failed. Contact the report owner for details.",
header_data=header_data,
url=url,
retry_attempt=attempt,
@@ -1652,14 +1835,17 @@ class BaseReportState:
max_attempts: int = self._report_schedule.retry_max_attempts
notification_content = NotificationContent(
name=sanitize_title(self._report_schedule.name),
- text=error_message,
+ text=(
+ "Report execution failed after retries. "
+ "Contact the report owner for details."
+ ),
header_data=header_data,
url=url,
retry_max_attempts=max_attempts,
)
self._send(notification_content, self._report_schedule.recipients)
- def _handle_retry_or_error(
+ def _handle_retry_or_error( # noqa: C901
self, error_message: str, original_exception: Exception
) -> bool:
"""
@@ -1674,6 +1860,10 @@ class BaseReportState:
if not retry_on_failure:
return False
+ if self._delivery_started:
+ self._reset_retry_counter()
+ return False
+
max_attempts: int = self._report_schedule.retry_max_attempts
# If a new crontab window has fired since the first failure, reset the
@@ -1687,7 +1877,11 @@ class BaseReportState:
# send the retry-failure notification *after* the attempt ran so
# the email reflects what happened, not what is about to be
# scheduled. ("You will receive an update after each retry.")
- if current_attempt > 0:
+ # Suppress the last per-attempt notice only when a final notice replaces it.
+ if current_attempt > 0 and (
+ current_attempt < max_attempts
+ or not self._report_schedule.send_failed_reports
+ ):
try:
self.send_retry_notification(
current_attempt, max_attempts, error_message
@@ -1741,8 +1935,13 @@ class BaseReportState:
last_working = ReportScheduleDAO.find_last_entered_working_log(
self._report_schedule
)
- if not last_working:
+ if not last_working and not self._report_schedule.execution_owner:
return False
+ # A worker can die between committing its claim and creating the log.
+ # In that gap, a previous execution's old log must not expire this claim.
+ entered_at = self._report_schedule.last_eval_dttm
+ if not self._report_schedule.execution_owner and last_working:
+ entered_at = last_working.end_dttm
working_timeout = self._report_schedule.working_timeout
if self._report_schedule.type == ReportScheduleType.REPORT:
# Same effective budget the execution enforces (global budget
@@ -1756,10 +1955,10 @@ class BaseReportState:
)
return (
working_timeout is not None
- and self._report_schedule.last_eval_dttm is not None
+ and entered_at is not None
and datetime.now(timezone.utc).replace(tzinfo=None)
- timedelta(seconds=working_timeout)
- > last_working.end_dttm
+ > entered_at
)
def next(self) -> None:
@@ -1812,6 +2011,7 @@ class ReportNotTriggeredErrorState(BaseReportState):
self._report_schedule, self._execution_id
).run()
if not triggered:
+ self._reset_retry_counter()
self.update_report_schedule_and_log(
ReportState.NOOP, error_message=message
)
@@ -1918,6 +2118,25 @@ class ReportWorkingState(BaseReportState):
def next(self) -> None:
if self.is_on_working_timeout():
+ with db.session.no_autoflush:
+ recovered = (
+ db.session.query(ReportSchedule)
+ .filter(
+ ReportSchedule.id == self._report_schedule.id,
+ ReportSchedule.last_state == ReportState.WORKING,
+ ReportSchedule.execution_owner
+ == self._report_schedule.execution_owner,
+ ReportSchedule.last_eval_dttm
+ == self._report_schedule.last_eval_dttm,
+ )
+ .update(
+ {ReportSchedule.execution_owner: None},
+ synchronize_session=False,
+ )
+ )
+ if recovered != 1:
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ raise ReportSchedulePreviousWorkingError()
last_working = ReportScheduleDAO.find_last_entered_working_log(
self._report_schedule
)
@@ -1997,11 +2216,14 @@ class ReportSuccessState(BaseReportState):
self._report_schedule, self._execution_id
).run()
if not triggered:
+ self._reset_retry_counter()
self.update_report_schedule_and_log(
ReportState.NOOP, error_message=message
)
return
except Exception as ex:
+ if self._handle_retry_or_error(str(ex), ex):
+ return
# Ensure the schedule always transitions out of WORKING to
# ERROR, even if sending the error notification itself fails —
# otherwise the schedule is stuck in WORKING until the working
@@ -2129,17 +2351,24 @@ class ReportScheduleStateMachine: # pylint: disable=too-few-public-methods
report_schedule: ReportSchedule,
scheduled_dttm: datetime,
report_execution_context: ReportExecutionContext | None = None,
+ execution_claim: ExecutionClaim | None = None,
):
self._execution_id = task_uuid
self._report_schedule = report_schedule
self._scheduled_dttm = scheduled_dttm
self._report_execution_context = report_execution_context
+ self._execution_claim = execution_claim
@transaction()
def run(self) -> None:
+ initial_state = (
+ self._execution_claim.initial_state
+ if self._execution_claim is not None
+ else self._report_schedule.last_state
+ )
for state_cls in self.states_cls:
- if (self._report_schedule.last_state is None and state_cls.initial) or (
- self._report_schedule.last_state in state_cls.current_states
+ if (initial_state is None and state_cls.initial) or (
+ initial_state in state_cls.current_states
):
state_cls(
self._report_schedule,
@@ -2159,38 +2388,69 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
- On Alerts uses related Command AlertCommand and sends configured notifications
"""
- def __init__(self, task_id: str, model_id: int, scheduled_dttm: datetime):
+ def __init__(
+ self,
+ task_id: str,
+ model_id: int,
+ scheduled_dttm: datetime,
+ *,
+ is_retry: bool = False,
+ expected_owner: str | None = None,
+ ):
self._model_id = model_id
self._model: Optional[ReportSchedule] = None
self._scheduled_dttm = scheduled_dttm
self._execution_id = UUID(task_id)
+ self._is_retry = is_retry
+ self._expected_owner = expected_owner
- def run(self) -> None:
+ def run(self) -> None: # noqa: C901
monotonic_started_at = time.monotonic()
report_execution_context: ReportExecutionContext | None = None
- owns_report_working_state = False
+ owns_working_state = False
try:
self.validate()
if not self._model:
raise ReportScheduleExecuteUnexpectedError()
- # Reports always run under an execution context; alerts join them
- # only when they deliver a rendered screenshot, so a blank/partial
- # capture fails closed instead of being delivered. Ownership and
- # terminal-error persistence remain report-only recovery semantics.
+ retries_enabled = feature_flag_manager.is_feature_enabled(
+ "ALERT_REPORTS_RETRY"
+ )
+ if self._is_retry and (
+ not retries_enabled
+ or not self._model.retry_on_failure
+ or self._model.last_state != ReportState.RETRYING
+ or normalize_window(self._model.retry_scheduled_dttm)
+ != normalize_window(self._scheduled_dttm)
+ ):
+ if not retries_enabled or not self._model.retry_on_failure:
+ cancelled = cancel_disabled_retry(
+ db.session,
+ self._model.id,
+ self._scheduled_dttm,
+ self._expected_owner,
+ retries_enabled=retries_enabled,
+ )
+ if cancelled:
+ logger.info(
+ "report_retry_cancelled report_schedule_id=%s "
+ "reason=retries_disabled",
+ self._model_id,
+ )
+ logger.info(
+ "report_retry_discarded report_schedule_id=%s execution_id=%s",
+ self._model_id,
+ self._execution_id,
+ )
+ return
+
+ # All scheduled executions share ownership and retry fencing.
if _should_build_execution_context(self._model):
# An invocation that enters on WORKING is a duplicate or stale
# recovery, not the owner that created the active row. Its state
# handler may terminalize a stale execution, but the command
# boundary must never infer ownership from a replayed UUID.
- owns_report_working_state = (
- self._model.type == ReportScheduleType.REPORT
- and self._model.last_state != ReportState.WORKING
- )
- total_seconds = resolve_report_execution_budget_seconds(
- app.config,
- working_timeout=self._model.working_timeout,
- )
+ total_seconds = _execution_budget_seconds(self._model)
deadline = ReportExecutionDeadline(
total_seconds=total_seconds,
started_at=monotonic_started_at,
@@ -2203,6 +2463,7 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
)
report_execution_context = ReportExecutionContext(
execution_id=self._execution_id,
+ attempt=self._model.retry_attempt + 1 if self._is_retry else 1,
report_schedule_id=self._model.id,
dashboard_id=self._model.dashboard_id,
chart_id=self._model.chart_id,
@@ -2210,13 +2471,19 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
deadline=deadline,
capture_reserve_seconds=float(
app.config["ALERT_REPORTS_EXECUTION_CAPTURE_RESERVE_SECONDS"]
- ),
+ )
+ if self._model.type == ReportScheduleType.REPORT
+ else 0.0,
delivery_reserve_seconds=float(
app.config["ALERT_REPORTS_EXECUTION_DELIVERY_RESERVE_SECONDS"]
- ),
+ )
+ if self._model.type == ReportScheduleType.REPORT
+ else 0.0,
cleanup_reserve_seconds=float(
app.config["ALERT_REPORTS_EXECUTION_CLEANUP_RESERVE_SECONDS"]
- ),
+ )
+ if self._model.type == ReportScheduleType.REPORT
+ else 0.0,
)
logger.info(
"report_execution_start %s total_budget_seconds=%.2f "
@@ -2262,11 +2529,39 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
self._execution_id,
report_execution_context,
).get_dashboard_urls()
+ execution_claim = None
+ if self._model.last_state != ReportState.WORKING:
+ execution_claim = claim_execution(
+ db.session,
+ self._model.id,
+ str(self._execution_id),
+ self._scheduled_dttm,
+ is_retry=self._is_retry,
+ expected_owner=self._expected_owner,
+ retries_enabled=feature_flag_manager.is_feature_enabled(
+ "ALERT_REPORTS_RETRY"
+ ),
+ stale_retry_seconds=app.config.get(
+ "ALERT_REPORTS_RETRY_MAX_DELAY_SECONDS", 3600
+ ),
+ )
+ if execution_claim is None:
+ logger.info(
+ "report_execution_not_claimed schedule_id=%s",
+ self._model.id,
+ )
+ return
+ owns_working_state = True
+ if report_execution_context is not None:
+ report_execution_context = replace(
+ report_execution_context, execution_claimed=True
+ )
ReportScheduleStateMachine(
self._execution_id,
self._model,
self._scheduled_dttm,
report_execution_context,
+ execution_claim,
).run()
elapsed_seconds: float = (
@@ -2279,11 +2574,7 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
self._execution_id,
)
except (CommandException, SoftTimeLimitExceeded) as ex:
- if (
- self._model
- and self._model.type == ReportScheduleType.REPORT
- and owns_report_working_state
- ):
+ if self._model and owns_working_state:
persist_owned_report_execution_terminal_error(
self._model.id,
self._execution_id,
@@ -2293,11 +2584,7 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
)
raise
except Exception as ex:
- if (
- self._model
- and self._model.type == ReportScheduleType.REPORT
- and owns_report_working_state
- ):
+ if self._model and owns_working_state:
persist_owned_report_execution_terminal_error(
self._model.id,
self._execution_id,
diff --git a/superset/commands/report/execution_claim.py b/superset/commands/report/execution_claim.py
new file mode 100644
index 00000000000..dcecd4ea758
--- /dev/null
+++ b/superset/commands/report/execution_claim.py
@@ -0,0 +1,199 @@
+# 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.
+"""Atomic admission for scheduled report executions and their retries."""
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy.exc import DBAPIError
+from sqlalchemy.orm import Session
+
+from superset.reports.models import ReportSchedule, ReportState
+
+logger = logging.getLogger(__name__)
+
+
+def normalize_window(value: datetime | None) -> datetime | None:
+ """Compare database timestamps and broker timestamps at UTC second precision."""
+ if value is None:
+ return None
+ if value.tzinfo is not None:
+ value = value.astimezone(timezone.utc)
+ return value.replace(tzinfo=None, microsecond=0)
+
+
+@dataclass(frozen=True)
+class ExecutionClaim:
+ """State observed by the worker which won the database claim."""
+
+ initial_state: str | None
+
+
+def cancel_disabled_retry(
+ session: Session,
+ schedule_id: int,
+ window: datetime,
+ expected_owner: str | None,
+ *,
+ retries_enabled: bool,
+) -> bool:
+ """Release a disabled retry only when the queued owner's window still matches."""
+ if not expected_owner:
+ return False
+ query = session.query(ReportSchedule).filter(
+ ReportSchedule.id == schedule_id,
+ ReportSchedule.last_state == ReportState.RETRYING,
+ ReportSchedule.execution_owner == expected_owner,
+ ReportSchedule.execution_window == normalize_window(window),
+ ReportSchedule.retry_scheduled_dttm == normalize_window(window),
+ )
+ if retries_enabled:
+ query = query.filter(~ReportSchedule.retry_on_failure)
+ matched = query.update(
+ {
+ ReportSchedule.last_state: ReportState.ERROR,
+ ReportSchedule.last_eval_dttm: datetime.now(timezone.utc).replace(
+ tzinfo=None
+ ),
+ ReportSchedule.retry_attempt: 0,
+ ReportSchedule.retry_scheduled_dttm: None,
+ },
+ synchronize_session=False,
+ )
+ # Retain owner/window to fence stale replays of the cancelled execution.
+ session.commit() # pylint: disable=consider-using-transaction
+ return matched == 1
+
+
+def claim_execution(
+ session: Session,
+ schedule_id: int,
+ execution_id: str,
+ window: datetime,
+ *,
+ is_retry: bool,
+ expected_owner: str | None,
+ retries_enabled: bool,
+ stale_retry_seconds: float,
+) -> ExecutionClaim | None:
+ """Claim work, treating database serialization losers as unclaimed."""
+ try:
+ return _claim_execution(
+ session,
+ schedule_id,
+ execution_id,
+ window,
+ is_retry=is_retry,
+ expected_owner=expected_owner,
+ retries_enabled=retries_enabled,
+ stale_retry_seconds=stale_retry_seconds,
+ )
+ except DBAPIError as ex:
+ session.rollback() # pylint: disable=consider-using-transaction
+ # PostgreSQL REPEATABLE READ can raise instead of returning zero rows
+ # when another claimant updates the snapshot we observed.
+ sqlstate = getattr(ex.orig, "sqlstate", None) or getattr(
+ ex.orig, "pgcode", None
+ )
+ if sqlstate not in {"40001", "40P01"}:
+ raise
+ logger.info(
+ "report_execution_claim_conflict schedule_id=%s "
+ "execution_id=%s sqlstate=%s",
+ schedule_id,
+ execution_id,
+ sqlstate,
+ )
+ return None
+
+
+def _claim_execution(
+ session: Session,
+ schedule_id: int,
+ execution_id: str,
+ window: datetime,
+ *,
+ is_retry: bool,
+ expected_owner: str | None,
+ retries_enabled: bool,
+ stale_retry_seconds: float,
+) -> ExecutionClaim | None:
+ """Atomically promote an eligible schedule to WORKING, committing ownership.
+
+ A conditional UPDATE is the claim, not the preceding read. A competing
+ worker observing the same row loses when any admission field changes.
+ Ownership survives terminal states to fence broker redelivery of a window.
+ """
+ schedule = (
+ session.query(ReportSchedule)
+ .filter_by(id=schedule_id)
+ .populate_existing()
+ .one_or_none()
+ )
+ if schedule is None or schedule.last_state == ReportState.WORKING:
+ return None
+ normalized = normalize_window(window)
+ assert normalized is not None
+ initial_state = schedule.last_state
+ if is_retry:
+ if (
+ not retries_enabled
+ or not schedule.retry_on_failure
+ or initial_state != ReportState.RETRYING
+ or not expected_owner
+ or schedule.execution_owner != expected_owner
+ or normalize_window(schedule.retry_scheduled_dttm) != normalized
+ ):
+ return None
+ else:
+ previous_window = normalize_window(schedule.execution_window)
+ if previous_window is not None and normalized <= previous_window:
+ return None
+ if initial_state == ReportState.RETRYING:
+ anchor = normalize_window(schedule.retry_scheduled_dttm)
+ touched = normalize_window(schedule.last_eval_dttm)
+ if anchor == normalized or (
+ touched is not None
+ and (datetime.utcnow() - touched).total_seconds() < stale_retry_seconds
+ ):
+ return None
+ matched = (
+ session.query(ReportSchedule)
+ .filter(
+ ReportSchedule.id == schedule_id,
+ ReportSchedule.last_state == initial_state,
+ ReportSchedule.last_eval_dttm == schedule.last_eval_dttm,
+ ReportSchedule.execution_owner == schedule.execution_owner,
+ ReportSchedule.execution_window == schedule.execution_window,
+ ReportSchedule.retry_on_failure == schedule.retry_on_failure,
+ ReportSchedule.retry_scheduled_dttm == schedule.retry_scheduled_dttm,
+ )
+ .update(
+ {
+ ReportSchedule.last_state: ReportState.WORKING,
+ ReportSchedule.last_eval_dttm: datetime.utcnow(),
+ ReportSchedule.execution_owner: execution_id,
+ ReportSchedule.execution_window: normalized,
+ },
+ synchronize_session=False,
+ )
+ )
+ session.commit() # pylint: disable=consider-using-transaction
+ if matched != 1:
+ return None
+ return ExecutionClaim(initial_state=initial_state)
diff --git a/superset/commands/report/update.py b/superset/commands/report/update.py
index 7b25ede3f47..329d8a63129 100644
--- a/superset/commands/report/update.py
+++ b/superset/commands/report/update.py
@@ -206,11 +206,5 @@ class UpdateReportScheduleCommand(UpdateMixin, BaseReportScheduleCommand):
msg = _("send_failed_reports requires retry_on_failure to be enabled")
exceptions.append(ValidationError({"send_failed_reports": [msg]}))
- # Retries are only supported for reports, not alerts.
- report_type = self._properties.get("type", self._model.type)
- if report_type == ReportScheduleType.ALERT and retry_enabled:
- msg = _("Retries are not supported for alerts")
- exceptions.append(ValidationError({"retry_on_failure": [msg]}))
-
if exceptions:
raise ReportScheduleInvalidError(exceptions=exceptions)
diff --git a/superset/migrations/versions/2026-09-16_00-00_93d1b4a76c02_add_report_execution_ownership.py b/superset/migrations/versions/2026-09-16_00-00_93d1b4a76c02_add_report_execution_ownership.py
new file mode 100644
index 00000000000..b08fba05804
--- /dev/null
+++ b/superset/migrations/versions/2026-09-16_00-00_93d1b4a76c02_add_report_execution_ownership.py
@@ -0,0 +1,55 @@
+# 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.
+
+"""Add report execution ownership fencing.
+
+Revision ID: 93d1b4a76c02
+Revises: 60f94cd6cd11
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "93d1b4a76c02"
+down_revision = "60f94cd6cd11"
+
+
+def upgrade() -> None:
+ """Add nullable admission fields without rewriting existing schedules."""
+ existing = {
+ column["name"]
+ for column in sa.inspect(op.get_bind()).get_columns("report_schedule")
+ }
+ with op.batch_alter_table("report_schedule") as batch_op:
+ for column in (
+ sa.Column("execution_owner", sa.String(36), nullable=True),
+ sa.Column("execution_window", sa.DateTime(), nullable=True),
+ ):
+ if column.name not in existing:
+ batch_op.add_column(column)
+
+
+def downgrade() -> None:
+ """Remove admission fields; retry configuration remains unchanged."""
+ existing = {
+ column["name"]
+ for column in sa.inspect(op.get_bind()).get_columns("report_schedule")
+ }
+ with op.batch_alter_table("report_schedule") as batch_op:
+ for name in ("execution_window", "execution_owner"):
+ if name in existing:
+ batch_op.drop_column(name)
diff --git a/superset/reports/models.py b/superset/reports/models.py
index 886f89bc52a..d6062683ce6 100644
--- a/superset/reports/models.py
+++ b/superset/reports/models.py
@@ -183,6 +183,8 @@ class ReportSchedule(AuditMixinNullable, ExtraJSONMixin, Model):
# Retry state — written by the execution engine, not user-configurable
retry_attempt = Column(Integer, default=0, nullable=False, server_default="0")
retry_scheduled_dttm = Column(DateTime, nullable=True)
+ execution_owner = Column(String(36), nullable=True)
+ execution_window = Column(DateTime, nullable=True)
extra: ReportScheduleExtra # type: ignore
diff --git a/superset/reports/notifications/base.py b/superset/reports/notifications/base.py
index 63b08d42247..4986fca7a4b 100644
--- a/superset/reports/notifications/base.py
+++ b/superset/reports/notifications/base.py
@@ -40,6 +40,7 @@ class NotificationContent:
retry_attempt: Optional[int] = None
retry_max_attempts: Optional[int] = None
include_cta: bool = True # include the call-to-action link back to Superset
+ is_editor_error: bool = False # diagnostic email restricted to schedule editors
@property
def has_attachments(self) -> bool:
diff --git a/superset/reports/notifications/email.py b/superset/reports/notifications/email.py
index 5f1687e872f..71998a4c984 100644
--- a/superset/reports/notifications/email.py
+++ b/superset/reports/notifications/email.py
@@ -148,12 +148,12 @@ class EmailNotification(BaseNotification): # pylint: disable=too-few-public-met
return parseaddr(current_app.config["SMTP_MAIL_FROM"])[1].split("@")[1]
def _error_template(self, text: str) -> str:
- # The error text is derived from exception messages that can embed
- # data-controlled content (e.g. crafted table/column names in a DB
- # error). Strip all HTML before interpolating it into the email body,
- # matching the sanitization applied to the normal content path.
- # pylint: disable=no-member
- safe_text = nh3.clean(text, tags=set(), attributes={})
+ # Only send_error's editor-only email may include sanitized diagnostics.
+ safe_text = (
+ nh3.clean(text, tags=set(), attributes={})
+ if self._content.is_editor_error
+ else __("Contact the report owner for error details.")
+ )
if self._content.include_cta:
return __(
"""
@@ -179,7 +179,7 @@ class EmailNotification(BaseNotification): # pylint: disable=too-few-public-met
max_attempts = self._content.retry_max_attempts
retries_remaining = (max_attempts or 0) - (attempt or 0)
# pylint: disable=no-member
- safe_text = nh3.clean(text, tags=set(), attributes={})
+ safe_text = __("Contact the report owner for error details.")
cta_tag = self._render_call_to_action_paragraph()
return textwrap.dedent(
@@ -215,7 +215,7 @@ class EmailNotification(BaseNotification): # pylint: disable=too-few-public-met
def _final_failure_template(self, text: str) -> str:
"""HTML body for the final-failure email after all retries are exhausted."""
# pylint: disable=no-member
- safe_text = nh3.clean(text, tags=set(), attributes={})
+ safe_text = __("Contact the report owner for error details.")
cta_tag = self._render_call_to_action_paragraph()
max_attempts = self._content.retry_max_attempts
diff --git a/superset/reports/notifications/slack_mixin.py b/superset/reports/notifications/slack_mixin.py
index 2d38e746492..a01f6aa31cf 100644
--- a/superset/reports/notifications/slack_mixin.py
+++ b/superset/reports/notifications/slack_mixin.py
@@ -103,7 +103,7 @@ Error: %(text)s
attempt=retry_attempt,
max=retry_max_attempts,
remaining=retries_remaining,
- text=text,
+ text=__("Contact the report owner for error details."),
)
if retry_max_attempts is not None:
return __(
@@ -118,7 +118,7 @@ Error: %(text)s
name=name,
description=description,
max=retry_max_attempts,
- text=text,
+ text=__("Contact the report owner for error details."),
)
return __(
"""*%(name)s*
@@ -129,7 +129,7 @@ Error: %(text)s
""",
name=name,
description=description,
- text=text,
+ text=__("Contact the report owner for error details."),
)
def _get_body(self, content: NotificationContent) -> str:
diff --git a/superset/reports/notifications/webhook.py b/superset/reports/notifications/webhook.py
index 52e5183797d..d43ec0f33e4 100644
--- a/superset/reports/notifications/webhook.py
+++ b/superset/reports/notifications/webhook.py
@@ -23,6 +23,7 @@ from urllib.parse import urlparse
import backoff
import requests
from flask import current_app
+from flask_babel import gettext as __
from requests.adapters import HTTPAdapter
from urllib3.connection import HTTPConnection, HTTPSConnection
from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
@@ -180,7 +181,13 @@ class WebhookNotification(BaseNotification):
content = {
"name": self._content.name,
"header": header_content,
- "text": self._content.text,
+ # NotificationContent.text carries failure diagnostics; successful
+ # tabular output uses embedded_data rather than this field.
+ "text": (
+ __("Contact the report owner for error details.")
+ if self._content.text
+ else self._content.text
+ ),
"description": self._content.description,
"url": self._content.url,
}
diff --git a/superset/reports/schemas.py b/superset/reports/schemas.py
index 1b911a6a398..6d3d41e989d 100644
--- a/superset/reports/schemas.py
+++ b/superset/reports/schemas.py
@@ -401,13 +401,6 @@ class ReportSchedulePostSchema(RetryFieldStripMixin, Schema):
]
}
)
- # Retry is only supported for reports, not alerts.
- if data.get("type") == ReportScheduleType.ALERT and data.get(
- "retry_on_failure"
- ):
- raise ValidationError(
- {"retry_on_failure": [_("Retries are not supported for alerts")]}
- )
class ReportScheduleSubscribeSchema(ReportSchedulePostSchema):
diff --git a/superset/tasks/scheduler.py b/superset/tasks/scheduler.py
index 0e1fef705b4..91b3ecdd4d8 100644
--- a/superset/tasks/scheduler.py
+++ b/superset/tasks/scheduler.py
@@ -135,10 +135,21 @@ def execute(
self: Task,
report_schedule_id: int,
scheduled_dttm_iso: str | None = None,
+ expected_owner: str | None = None,
) -> None:
stats_logger: BaseStatsLogger = current_app.config["STATS_LOGGER"]
stats_logger.incr("reports.execute")
+ if scheduled_dttm_iso is not None and not expected_owner:
+ stats_logger.incr("reports.execute.legacy_retry_discarded")
+ logger.warning(
+ "report_retry_discarded report_schedule_id=%s execution_id=%s "
+ "reason=missing_execution_owner action=rerun_after_worker_upgrade",
+ report_schedule_id,
+ self.request.id,
+ )
+ return
+
task_id = None
try:
task_id = execute.request.id
@@ -165,6 +176,8 @@ def execute(
task_id,
report_schedule_id,
scheduled_dttm,
+ is_retry=scheduled_dttm_iso is not None,
+ expected_owner=expected_owner,
).run()
except SoftTimeLimitExceeded:
stats_logger.incr("reports.execute.celery_soft_timeout")
diff --git a/superset/utils/report_execution.py b/superset/utils/report_execution.py
index 7f91b48e41c..ad924878b1a 100644
--- a/superset/utils/report_execution.py
+++ b/superset/utils/report_execution.py
@@ -21,6 +21,7 @@ from __future__ import annotations
import logging
import time
from collections.abc import Callable, Mapping
+from copy import copy
from dataclasses import dataclass, field
from typing import Any
from uuid import UUID
@@ -190,6 +191,7 @@ class ReportExecutionContext:
capture_reserve_seconds: float = 0.0
delivery_reserve_seconds: float = 0.0
cleanup_reserve_seconds: float = 0.0
+ execution_claimed: bool = False
_capture_rejection_reasons: list[str] = field(
default_factory=list,
init=False,
@@ -213,6 +215,19 @@ class ReportExecutionContext:
"execution budget"
)
+ def with_deadline(
+ self, deadline: ReportExecutionDeadline
+ ) -> "ReportExecutionContext":
+ """Share execution state while bounding a phase with a separate deadline.
+
+ dataclasses.replace resets init=False fields, including sticky capture
+ rejection. A shallow copy must retain those shared mutable containers.
+ """
+ context = copy(self)
+ object.__setattr__(context, "deadline", deadline)
+ context.__post_init__()
+ return context
+
@property
def log_context(self) -> str:
"""Return stable key/value identifiers for plain-text log formatters."""
diff --git a/tests/integration_tests/reports/commands_tests.py b/tests/integration_tests/reports/commands_tests.py
index 6490ab4b351..1efd1706743 100644
--- a/tests/integration_tests/reports/commands_tests.py
+++ b/tests/integration_tests/reports/commands_tests.py
@@ -68,7 +68,10 @@ from superset.commands.report.execute import (
BaseReportState,
)
from superset.commands.report.log_prune import AsyncPruneReportScheduleLogCommand
-from superset.daos.report import ReportScheduleDAO
+from superset.daos.report import (
+ REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER,
+ ReportScheduleDAO,
+)
from superset.exceptions import SupersetException
from superset.key_value.models import KeyValueEntry
from superset.models.core import Database
@@ -1293,7 +1296,15 @@ def test_email_chart_report_schedule_with_text(
# Assert logs are correct
assert_log(ReportState.SUCCESS)
- # test with date type.
+ # Redelivering the completed window must not send a second email.
+ with freeze_time("2020-01-01T00:00:00Z"):
+ AsyncExecuteReportScheduleCommand(
+ str(uuid4()), create_report_email_chart_with_text.id, datetime.utcnow()
+ ).run()
+ email_mock.assert_called_once()
+ assert_log(ReportState.SUCCESS)
+
+ # Test date columns in a new scheduled window.
dt = datetime(2022, 1, 1).replace(tzinfo=timezone.utc)
ts = datetime.timestamp(dt) * 1000
response.read.return_value = json.dumps(
@@ -1313,7 +1324,7 @@ def test_email_chart_report_schedule_with_text(
}
).encode("utf-8")
- with freeze_time("2020-01-01T00:00:00Z"):
+ with freeze_time("2020-01-01T00:01:00Z"):
AsyncExecuteReportScheduleCommand(
TEST_ID, create_report_email_chart_with_text.id, datetime.utcnow()
).run()
@@ -1780,12 +1791,15 @@ def test_slack_chart_report_schedule_with_errors(
SlackApiError(message="foo", response="bar"),
]
- for idx, er in enumerate(slack_errors): # noqa: B007
+ scheduled_dttm = datetime.utcnow()
+ for idx, er in enumerate(slack_errors):
web_client_mock.side_effect = [SlackApiError(None, None), er]
with pytest.raises(ReportScheduleClientErrorsException):
AsyncExecuteReportScheduleCommand(
- TEST_ID, create_report_slack_chart.id, datetime.utcnow()
+ str(uuid4()),
+ create_report_slack_chart.id,
+ scheduled_dttm + timedelta(minutes=idx),
).run()
db.session.commit()
@@ -2825,19 +2839,24 @@ def test_readiness_timeout_retries_terminal_persistence_and_allows_next_schedule
)
create_report_email_chart.last_state = ReportState.SUCCESS
db.session.commit()
+ scheduled_dttm = datetime.utcnow()
with pytest.raises(ReportScheduleScreenshotFailedError):
AsyncExecuteReportScheduleCommand(
TEST_ID,
create_report_email_chart.id,
- datetime.utcnow(),
+ scheduled_dttm,
).run()
assert terminal_write_failed
db.session.refresh(create_report_email_chart)
timed_out_log = (
db.session.query(ReportExecutionLog)
- .filter(ReportExecutionLog.uuid == UUID(TEST_ID))
+ .filter(
+ ReportExecutionLog.uuid == UUID(TEST_ID),
+ ReportExecutionLog.error_message
+ != REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER,
+ )
.one()
)
assert timed_out_log.state == ReportState.ERROR
@@ -2847,7 +2866,7 @@ def test_readiness_timeout_retries_terminal_persistence_and_allows_next_schedule
# MySQL's metadata schema can store these values with one-second precision.
assert timed_out_log.end_dttm >= timed_out_log.start_dttm
assert create_report_email_chart.last_state == ReportState.ERROR
- email_mock.assert_not_called()
+ email_mock.assert_called_once()
assert any(
"report_execution_terminal" in record.message
and TEST_ID in record.message
@@ -2862,7 +2881,7 @@ def test_readiness_timeout_retries_terminal_persistence_and_allows_next_schedule
AsyncExecuteReportScheduleCommand(
next_execution_id,
create_report_email_chart.id,
- datetime.utcnow(),
+ scheduled_dttm + timedelta(minutes=1),
).run()
db.session.refresh(create_report_email_chart)
@@ -3215,15 +3234,16 @@ def test_retry_on_failure_schedules_retry(
@patch("superset.commands.report.execute.ReportNotTriggeredErrorState._schedule_retry")
@patch("superset.commands.report.execute.BaseReportState.send_retry_notification")
@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
+@pytest.mark.parametrize("send_failed_reports", [False, True])
def test_retry_exhausted_transitions_to_error(
screenshot_mock: Mock,
retry_notification_mock: Mock,
schedule_retry_mock: Mock,
+ send_failed_reports: bool,
) -> None:
"""
ExecuteReport Command: when all retries are exhausted the state transitions
- to ERROR, the retry counter is reset, and the retry notification is sent
- for the final attempt.
+ to ERROR and exactly one recipient failure notice is selected.
"""
chart = db.session.query(Slice).first()
report_schedule = create_report_notification(
@@ -3232,26 +3252,36 @@ def test_retry_exhausted_transitions_to_error(
retry_on_failure=True,
retry_max_attempts=2,
retry_notify_owners=False,
- retry_notify_recipients=False,
+ retry_notify_recipients=True,
+ send_failed_reports=send_failed_reports,
)
# Pre-set retry_attempt to the max so the next execution exhausts retries.
# Use the same timestamp for both so _is_retry_window_stale() returns False.
- # Truncate microseconds — MySQL DateTime columns drop them, which would make
- # the round-tripped value differ from the in-memory one.
# Truncate microseconds — MySQL DATETIME columns drop them, causing
# _is_retry_window_stale() to see a mismatch after DB round-trip.
scheduled_dttm = datetime.now(tz=timezone.utc).replace(tzinfo=None, microsecond=0)
report_schedule.retry_attempt = 2
report_schedule.retry_scheduled_dttm = scheduled_dttm
report_schedule.last_state = ReportState.RETRYING
+ previous_owner = str(uuid4())
+ report_schedule.execution_owner = previous_owner
db.session.commit()
try:
screenshot_mock.side_effect = Exception("screenshot failed")
- with pytest.raises(Exception, match="screenshot failed"):
+ with (
+ patch(
+ "superset.commands.report.execute.BaseReportState.send_final_failure_report"
+ ) as final_failure_mock,
+ pytest.raises(Exception, match="screenshot failed"),
+ ):
AsyncExecuteReportScheduleCommand(
- TEST_ID, report_schedule.id, scheduled_dttm
+ TEST_ID,
+ report_schedule.id,
+ scheduled_dttm,
+ is_retry=True,
+ expected_owner=previous_owner,
).run()
db.session.refresh(report_schedule)
@@ -3260,9 +3290,13 @@ def test_retry_exhausted_transitions_to_error(
assert report_schedule.retry_attempt == 0
# No further retry should have been scheduled
schedule_retry_mock.assert_not_called()
- # Retry notification sent for the exhausted attempt (attempt 2 of 2)
- # The error message is wrapped by the screenshot layer, so use ANY.
- retry_notification_mock.assert_called_once_with(2, 2, ANY)
+ error_message = "Failed taking a screenshot screenshot failed"
+ if send_failed_reports:
+ retry_notification_mock.assert_not_called()
+ final_failure_mock.assert_called_once_with(error_message)
+ else:
+ retry_notification_mock.assert_called_once_with(2, 2, error_message)
+ final_failure_mock.assert_not_called()
finally:
cleanup_report_schedule(report_schedule)
@@ -3295,6 +3329,8 @@ def test_send_failed_reports_sends_to_recipients(
report_schedule.retry_attempt = 1
report_schedule.retry_scheduled_dttm = scheduled_dttm
report_schedule.last_state = ReportState.RETRYING
+ previous_owner = str(uuid4())
+ report_schedule.execution_owner = previous_owner
db.session.commit()
try:
@@ -3302,7 +3338,11 @@ def test_send_failed_reports_sends_to_recipients(
with pytest.raises(Exception, match="screenshot failed"):
AsyncExecuteReportScheduleCommand(
- TEST_ID, report_schedule.id, scheduled_dttm
+ TEST_ID,
+ report_schedule.id,
+ scheduled_dttm,
+ is_retry=True,
+ expected_owner=previous_owner,
).run()
# send_final_failure_report should have been called
@@ -3338,6 +3378,8 @@ def test_retrying_state_schedules_another_retry(
report_schedule.last_state = ReportState.RETRYING
report_schedule.retry_attempt = 1
report_schedule.retry_scheduled_dttm = scheduled_dttm
+ previous_owner = str(uuid4())
+ report_schedule.execution_owner = previous_owner
db.session.commit()
try:
@@ -3345,7 +3387,11 @@ def test_retrying_state_schedules_another_retry(
# Should not raise — still within retry budget
AsyncExecuteReportScheduleCommand(
- TEST_ID, report_schedule.id, scheduled_dttm
+ TEST_ID,
+ report_schedule.id,
+ scheduled_dttm,
+ is_retry=True,
+ expected_owner=previous_owner,
).run()
db.session.refresh(report_schedule)
@@ -3423,13 +3469,19 @@ def test_retry_notify_owners_sends_notification(
report_schedule.last_state = ReportState.RETRYING
report_schedule.retry_attempt = 1
report_schedule.retry_scheduled_dttm = scheduled_dttm
+ previous_owner = str(uuid4())
+ report_schedule.execution_owner = previous_owner
db.session.commit()
try:
screenshot_mock.side_effect = Exception("screenshot failed")
AsyncExecuteReportScheduleCommand(
- TEST_ID, report_schedule.id, scheduled_dttm
+ TEST_ID,
+ report_schedule.id,
+ scheduled_dttm,
+ is_retry=True,
+ expected_owner=previous_owner,
).run()
# Notification sent for the failed retry (attempt 1 of 3)
@@ -3514,6 +3566,8 @@ def test_success_after_retry_clears_retry_state(
report_schedule.last_state = ReportState.RETRYING
report_schedule.retry_attempt = 2
report_schedule.retry_scheduled_dttm = scheduled_dttm
+ previous_owner = str(uuid4())
+ report_schedule.execution_owner = previous_owner
db.session.commit()
try:
@@ -3521,7 +3575,11 @@ def test_success_after_retry_clears_retry_state(
screenshot_mock.return_value = SCREENSHOT_FILE
AsyncExecuteReportScheduleCommand(
- TEST_ID, report_schedule.id, scheduled_dttm
+ TEST_ID,
+ report_schedule.id,
+ scheduled_dttm,
+ is_retry=True,
+ expected_owner=previous_owner,
).run()
db.session.refresh(report_schedule)
diff --git a/tests/integration_tests/reports/execution_claim_tests.py b/tests/integration_tests/reports/execution_claim_tests.py
new file mode 100644
index 00000000000..cc22be4be20
--- /dev/null
+++ b/tests/integration_tests/reports/execution_claim_tests.py
@@ -0,0 +1,32 @@
+# 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.
+
+"""Run shared connection-level regressions in the PostgreSQL CI matrix too."""
+
+import pytest
+
+from tests.unit_tests.commands.report.execution_claim_test import (
+ claim_engine as claim_engine,
+ sessions as sessions,
+ test_competing_workers_have_one_winner as test_claim_race,
+ test_recovery_can_fence_worker_during_transport as test_delivery_fencing,
+ test_terminal_fallback_persists_without_working_log as test_terminal_fallback,
+)
+
+__all__ = ["test_claim_race", "test_delivery_fencing", "test_terminal_fallback"]
+
+pytestmark = pytest.mark.usefixtures("app_context")
diff --git a/tests/unit_tests/commands/report/execute_test.py b/tests/unit_tests/commands/report/execute_test.py
index a5e7c710dbf..bfc5ba63917 100644
--- a/tests/unit_tests/commands/report/execute_test.py
+++ b/tests/unit_tests/commands/report/execute_test.py
@@ -47,6 +47,7 @@ from superset.commands.report.exceptions import (
)
from superset.commands.report.execute import (
_should_build_execution_context,
+ AsyncExecuteReportScheduleCommand,
BaseReportState,
log_report_delivery_phase,
persist_owned_report_execution_terminal_error,
@@ -69,12 +70,15 @@ from superset.reports.models import (
ReportState,
)
from superset.reports.notifications.base import BaseNotification, NotificationContent
+from superset.reports.notifications.email import EmailNotification
from superset.reports.notifications.exceptions import (
+ NotificationError,
NotificationParamException,
SlackV1NotificationError,
)
from superset.reports.notifications.slack import SlackNotification
from superset.reports.notifications.slack_channel_resolver import _match_slack_channel
+from superset.reports.notifications.slack_mixin import SlackMixin
from superset.subjects.types import SubjectType
from superset.utils.core import HeaderDataType
from superset.utils.report_execution import (
@@ -92,6 +96,476 @@ from superset.utils.slack import (
from tests.integration_tests.conftest import with_feature_flags
+@pytest.mark.parametrize("final", [False, True])
+def test_rendered_retry_messages_do_not_include_diagnostics(mocker, final):
+ content = NotificationContent(
+ name="Report",
+ header_data={},
+ url="https://example.com",
+ text="private smtp diagnostic: confidential@example.com",
+ retry_attempt=None if final else 1,
+ retry_max_attempts=3,
+ )
+ email = EmailNotification(mocker.Mock(spec=ReportRecipients), content)
+ for body in (email._get_content().body, SlackMixin()._get_body(content)):
+ assert "confidential@example.com" not in body
+ assert "private smtp" not in body
+ assert "report owner" in body
+
+
+@pytest.mark.parametrize("schedule_type", list(ReportScheduleType))
+def test_exhausting_retry_sends_only_final_notice(mocker, schedule_type):
+ state = _make_state_instance(mocker, BaseReportState, schedule_type=schedule_type)
+ state._report_schedule.retry_on_failure = True
+ state._report_schedule.retry_attempt = 3
+ state._report_schedule.send_failed_reports = True
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=True,
+ )
+ retry_notice = mocker.patch.object(state, "send_retry_notification")
+ final_notice = mocker.patch.object(state, "send_final_failure_report")
+ retry = mocker.patch.object(state, "_schedule_retry")
+ assert not state._handle_retry_or_error("failure", RuntimeError("failure"))
+ retry_notice.assert_not_called()
+ final_notice.assert_called_once_with("failure")
+ retry.assert_not_called()
+
+
+@pytest.mark.parametrize("schedule_type", list(ReportScheduleType))
+@pytest.mark.parametrize("send_failed_reports", [False, True])
+def test_last_retry_preserves_opted_in_recipient_notice(
+ mocker,
+ schedule_type,
+ send_failed_reports,
+):
+ state = _make_state_instance(
+ mocker,
+ ReportNotTriggeredErrorState,
+ schedule_type=schedule_type,
+ last_state=ReportState.RETRYING,
+ )
+ schedule = state._report_schedule
+ schedule.retry_on_failure = True
+ schedule.retry_attempt = schedule.retry_max_attempts = 1
+ schedule.retry_notify_owners = False
+ schedule.retry_notify_recipients = True
+ schedule.send_failed_reports = send_failed_reports
+ schedule.recipients = [mocker.Mock(spec=ReportRecipients)]
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=True,
+ )
+ mocker.patch(
+ "superset.commands.report.execute.AlertCommand.run", return_value=(True, None)
+ )
+ mocker.patch.object(state, "_is_retry_window_stale", return_value=False)
+ mocker.patch.object(state, "send", side_effect=RuntimeError("capture failed"))
+ mocker.patch.object(state, "update_report_schedule_and_log")
+ mocker.patch.object(state, "is_in_error_grace_period", return_value=True)
+ mocker.patch.object(state, "_get_log_data", return_value={})
+ mocker.patch.object(state, "_get_url", return_value="https://example.com")
+ send = mocker.patch.object(state, "_send")
+ final_notice = mocker.patch.object(state, "send_final_failure_report")
+ retry = mocker.patch.object(state, "_schedule_retry")
+
+ with pytest.raises(RuntimeError, match="capture failed"):
+ state.next()
+
+ if send_failed_reports:
+ final_notice.assert_called_once_with("capture failed")
+ send.assert_not_called()
+ else:
+ final_notice.assert_not_called()
+ send.assert_called_once()
+ content, recipients = send.call_args.args
+ assert recipients == schedule.recipients
+ assert content.retry_attempt == content.retry_max_attempts == 1
+ retry.assert_not_called()
+ assert schedule.retry_attempt == 0
+
+
+@pytest.mark.parametrize("age,expired", [(10, False), (4000, True)])
+def test_working_claim_timeout_does_not_require_its_log(mocker, age, expired):
+ state = _make_state_instance(
+ mocker, ReportWorkingState, schedule_type=ReportScheduleType.REPORT
+ )
+ state._report_schedule.execution_owner = str(state._execution_id)
+ state._report_schedule.last_eval_dttm = datetime.utcnow() - timedelta(seconds=age)
+ previous_log = mocker.Mock(end_dttm=datetime.utcnow() - timedelta(days=1))
+ mocker.patch(
+ "superset.commands.report.execute.ReportScheduleDAO.find_last_entered_working_log",
+ return_value=previous_log,
+ )
+ assert state.is_on_working_timeout() is expired
+
+
+@pytest.mark.parametrize("schedule_type", list(ReportScheduleType))
+def test_rejected_capture_enters_whole_execution_retry(
+ mocker: MockerFixture,
+ schedule_type: ReportScheduleType,
+) -> None:
+ state = _make_notification_state(mocker, schedule_type=schedule_type)
+ state._report_schedule.retry_on_failure = True
+ state._report_schedule.retry_max_attempts = 3
+ state._report_schedule.retry_attempt = 0
+ state._report_schedule.retry_scheduled_dttm = None
+ state._report_schedule.send_failed_reports = False
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=True,
+ )
+ mocker.patch.object(state, "_is_retry_window_stale", return_value=False)
+ update_log = mocker.patch.object(state, "update_report_schedule_and_log")
+ schedule_retry = mocker.patch.object(state, "_schedule_retry")
+
+ error = ReportScheduleScreenshotFailedError("blank capture rejected")
+
+ assert state._handle_retry_or_error(str(error), error) is True
+ assert state._report_schedule.retry_attempt == 1
+ update_log.assert_called_once_with(
+ ReportState.RETRYING,
+ error_message=str(error),
+ )
+ schedule_retry.assert_called_once_with(60)
+
+
+@pytest.mark.parametrize(
+ "state_cls", [ReportNotTriggeredErrorState, ReportSuccessState]
+)
+@pytest.mark.parametrize("failure_phase", ["capture", "query"])
+def test_alert_pre_delivery_failure_schedules_retry(
+ mocker: MockerFixture, state_cls: type, failure_phase: str
+) -> None:
+ """Initial and previously successful alerts retry pre-delivery failures."""
+ state = _make_state_instance(mocker, state_cls)
+ state._report_schedule.retry_on_failure = True
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=True,
+ )
+ mocker.patch.object(state, "is_in_grace_period", return_value=False)
+ update = mocker.patch.object(state, "update_report_schedule_and_log")
+ mocker.patch.object(state, "is_in_error_grace_period", return_value=True)
+ query = mocker.patch(
+ "superset.commands.report.execute.AlertCommand"
+ ).return_value.run
+ query.return_value = (True, "triggered")
+ send = mocker.patch.object(state, "send")
+ error = ReportScheduleScreenshotFailedError("invalid capture")
+ if failure_phase == "capture":
+ send.side_effect = error
+ else:
+ query.side_effect = error
+ retry = mocker.patch.object(state, "_schedule_retry")
+ notify = mocker.patch.object(state, "send_error")
+
+ state.next()
+
+ retry.assert_called_once_with(60)
+ assert state._report_schedule.retry_attempt == 1
+ update.assert_called_with(ReportState.RETRYING, error_message=str(error))
+ notify.assert_not_called()
+ if failure_phase == "query":
+ send.assert_not_called()
+
+
+@pytest.mark.parametrize("triggered", [False, True])
+def test_alert_retry_reevaluates_condition_and_clears_retry_state(
+ mocker: MockerFixture, triggered: bool
+) -> None:
+ """A recovered alert sends only if its freshly evaluated condition still fires."""
+ state = _make_state_instance(
+ mocker, ReportNotTriggeredErrorState, last_state=ReportState.RETRYING
+ )
+ state._report_schedule.retry_on_failure = True
+ state._report_schedule.retry_attempt = 1
+ state._report_schedule.retry_scheduled_dttm = state._scheduled_dttm
+ query = mocker.patch(
+ "superset.commands.report.execute.AlertCommand"
+ ).return_value.run
+ query.return_value = (triggered, "condition cleared")
+ send = mocker.patch.object(state, "send")
+ update = mocker.patch.object(state, "update_report_schedule_and_log")
+
+ state.next()
+
+ query.assert_called_once()
+ assert send.call_count == int(triggered)
+ assert state._report_schedule.retry_attempt == 0
+ assert state._report_schedule.retry_scheduled_dttm is None
+ assert update.call_args.args[0] == (
+ ReportState.SUCCESS if triggered else ReportState.NOOP
+ )
+
+
+@pytest.mark.parametrize("enabled,opted_in", [(False, True), (True, False)])
+def test_alert_retry_requires_flag_and_schedule_opt_in(
+ mocker: MockerFixture, enabled: bool, opted_in: bool
+) -> None:
+ """Enabling alert support must not opt existing schedules in implicitly."""
+ state = _make_state_instance(mocker, BaseReportState)
+ state._report_schedule.retry_on_failure = opted_in
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=enabled,
+ )
+ retry = mocker.patch.object(state, "_schedule_retry")
+ assert not state._handle_retry_or_error("failure", RuntimeError("failure"))
+ retry.assert_not_called()
+
+
+def test_whole_execution_retry_preserves_report_task_timeouts(
+ mocker: MockerFixture,
+) -> None:
+ state = _make_notification_state(mocker)
+ state._report_schedule.id = 11
+ state._report_schedule.working_timeout = 540
+ state._scheduled_dttm = datetime(2026, 9, 15, 0, 0)
+ execute_task = mocker.patch("superset.tasks.scheduler.execute")
+ timeout_options = mocker.patch(
+ "superset.commands.report.execute.get_report_task_timeout_options",
+ return_value={"soft_time_limit": 540, "time_limit": 570},
+ )
+
+ state._schedule_retry(60)
+
+ timeout_options.assert_called_once()
+ execute_task.apply_async.assert_called_once_with(
+ (11, "2026-09-15T00:00:00", str(state._execution_id)),
+ countdown=60,
+ soft_time_limit=540,
+ time_limit=570,
+ )
+
+
+@pytest.mark.parametrize("last_state", [ReportState.SUCCESS, ReportState.RETRYING])
+@pytest.mark.parametrize("enabled", [False, True])
+def test_old_retry_is_discarded_before_state_machine(
+ mocker: MockerFixture,
+ last_state: ReportState,
+ enabled: bool,
+) -> None:
+ """An expired retry must not execute after a newer window starts or succeeds."""
+ command = AsyncExecuteReportScheduleCommand(
+ str(uuid4()),
+ 11,
+ datetime(2026, 9, 15),
+ is_retry=True,
+ )
+ model = mocker.Mock(spec=ReportSchedule)
+ model.retry_on_failure = True
+ model.last_state = last_state
+ model.retry_scheduled_dttm = datetime(2026, 9, 16 if enabled else 15)
+ command._model = model
+ mocker.patch.object(command, "validate")
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=enabled,
+ )
+ machine = mocker.patch(
+ "superset.commands.report.execute.ReportScheduleStateMachine"
+ )
+ command.run()
+ machine.assert_not_called()
+
+
+@pytest.mark.parametrize("schedule_type", list(ReportScheduleType))
+def test_partial_delivery_does_not_retry_whole_execution(
+ mocker: MockerFixture, schedule_type: ReportScheduleType
+) -> None:
+ """Even a retry-enabled schedule must not replay a successful recipient."""
+ state = _make_state_instance(mocker, BaseReportState)
+ state._report_schedule.type = schedule_type
+ state._report_schedule.retry_on_failure = True
+ state._report_schedule.retry_attempt = 1
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=True,
+ )
+ retry = mocker.patch.object(state, "_schedule_retry")
+ email, slack = mocker.Mock(), mocker.Mock()
+ slack.send.side_effect = NotificationError("Slack failed")
+ mocker.patch(
+ "superset.commands.report.execute.create_notification",
+ side_effect=[email, slack],
+ )
+ mocker.patch.dict(
+ "superset.commands.report.execute.app.config",
+ {
+ "ALERT_REPORTS_NOTIFICATION_DRY_RUN": False,
+ },
+ )
+ with pytest.raises(ReportScheduleSystemErrorsException):
+ state._send(
+ NotificationContent(
+ name="report", header_data=state._get_log_data(), text="content"
+ ),
+ [mocker.Mock(spec=ReportRecipients), mocker.Mock(spec=ReportRecipients)],
+ )
+ email.send.assert_called_once()
+ slack.send.assert_called_once()
+ assert not state._handle_retry_or_error("SMTP recipient refused", RuntimeError())
+ retry.assert_not_called()
+ assert state._report_schedule.retry_attempt == 0
+
+
+@pytest.mark.parametrize("final", [True, False])
+def test_retry_recipient_notifications_redact_provider_errors(
+ mocker: MockerFixture,
+ final: bool,
+) -> None:
+ """Provider diagnostics must not disclose one recipient to another."""
+ state = _make_state_instance(mocker, BaseReportState)
+ state._report_schedule.retry_notify_recipients = True
+ state._report_schedule.recipients = [mocker.Mock(spec=ReportRecipients)]
+ mocker.patch.object(state, "_get_log_data", return_value={})
+ mocker.patch.object(state, "_get_url", return_value="https://example.com")
+ send = mocker.patch.object(state, "_send")
+ if final:
+ state.send_final_failure_report("confidential diagnostic")
+ else:
+ state.send_retry_notification(1, 3, "confidential diagnostic")
+ assert "confidential" not in send.call_args.args[0].text
+ assert "diagnostic" not in send.call_args.args[0].text
+
+
+def test_fenced_worker_cannot_start_delivery(mocker: MockerFixture) -> None:
+ """A worker which loses ownership cannot send even a valid artifact."""
+ from dataclasses import replace
+
+ state = _make_notification_state(mocker)
+ state._report_execution_context = replace(
+ _active_report_context(), execution_claimed=True
+ )
+ query = mocker.patch(
+ "superset.commands.report.execute.db.session.query"
+ ).return_value.filter.return_value
+ query.update.return_value = 0
+ send = mocker.patch.object(state, "_send_notification")
+ content = NotificationContent(
+ name="report",
+ header_data=state._get_log_data(),
+ screenshots=[b"captured image"],
+ )
+ with pytest.raises(ReportSchedulePreviousWorkingError):
+ state._send(content, [mocker.Mock(spec=ReportRecipients)])
+ send.assert_not_called()
+
+
+def test_delivery_releases_ownership_lock_before_transport(
+ mocker: MockerFixture,
+) -> None:
+ """The CAS transaction must finish before each recipient's network I/O."""
+ from dataclasses import replace
+
+ state = _make_notification_state(mocker)
+ state._report_execution_context = replace(
+ _active_report_context(), execution_claimed=True
+ )
+ session = mocker.patch("superset.commands.report.execute.db.session")
+ session.query.return_value.filter.return_value.update.return_value = 1
+ events = []
+ session.commit.side_effect = lambda: events.append("commit")
+ mocker.patch.object(
+ state, "_send_notification", side_effect=lambda *_: events.append("send")
+ )
+ content = NotificationContent(
+ name="report", text="failure", header_data=state._get_log_data()
+ )
+ state._send(content, [mocker.Mock(spec=ReportRecipients) for _ in range(2)])
+ assert events == ["commit", "send", "commit", "send"]
+ session.query.return_value.filter.return_value.with_for_update.assert_not_called()
+
+
+@pytest.mark.parametrize("schedule_type", list(ReportScheduleType))
+@pytest.mark.parametrize("lost_owner", [False, True])
+def test_claimed_execution_fallback_without_working_log(
+ mocker: MockerFixture, schedule_type: ReportScheduleType, lost_owner: bool
+) -> None:
+ """A rolled-back log cannot leave a durable owned claim in WORKING."""
+ from dataclasses import replace
+
+ context = replace(_active_report_context(), execution_claimed=True)
+ schedule = mocker.Mock(spec=ReportSchedule)
+ schedule.type = schedule_type
+ schedule.execution_window = datetime.utcnow()
+ session = mocker.patch("superset.commands.report.execute.db.session")
+ session.query.return_value.filter_by.return_value.one_or_none.return_value = (
+ schedule
+ )
+ session.query.return_value.filter.return_value.update.return_value = (
+ 0 if lost_owner else 1
+ )
+ state = mocker.patch(
+ "superset.commands.report.execute.BaseReportState"
+ ).return_value
+ state.is_in_error_grace_period.return_value = False
+ assert persist_owned_report_execution_terminal_error(
+ 11, context.execution_id, "failure", "RuntimeError", context
+ ) is (not lost_owner)
+ if lost_owner:
+ state.create_log.assert_not_called()
+ state.send_error.assert_not_called()
+ else:
+ assert state.create_log.call_args_list[0] == mocker.call(
+ "failure", log_state=ReportState.ERROR
+ )
+ state.send_error.assert_called_once()
+
+
+@pytest.mark.parametrize("schedule_type", list(ReportScheduleType))
+@pytest.mark.parametrize(
+ "error",
+ [
+ RuntimeError("escaped"),
+ SoftTimeLimitExceeded(),
+ ReportScheduleStateNotFoundError(),
+ ],
+)
+def test_command_terminalizes_claimed_alert_and_report_on_escape(
+ mocker: MockerFixture, schedule_type: ReportScheduleType, error: Exception
+) -> None:
+ """Both schedule types must reach the fallback after committing a claim."""
+ from superset.commands.report.execution_claim import ExecutionClaim
+
+ command = AsyncExecuteReportScheduleCommand(str(uuid4()), 11, datetime.utcnow())
+ command._model = ReportSchedule(
+ id=11,
+ name="fallback",
+ type=schedule_type,
+ crontab="* * * * *",
+ last_state=ReportState.NOOP,
+ working_timeout=600,
+ )
+ mocker.patch.object(command, "validate")
+ mocker.patch(
+ "superset.commands.report.execute.get_executor", return_value=(None, "admin")
+ )
+ mocker.patch(
+ "superset.commands.report.execute.security_manager.find_user", return_value=None
+ )
+ mocker.patch(
+ "superset.commands.report.execute.claim_execution",
+ return_value=ExecutionClaim(ReportState.NOOP),
+ )
+ mocker.patch(
+ "superset.commands.report.execute.ReportScheduleStateMachine"
+ ).return_value.run.side_effect = error
+ persist = mocker.patch(
+ "superset.commands.report.execute.persist_owned_report_execution_terminal_error"
+ )
+ expected_error = (
+ ReportScheduleUnexpectedError
+ if isinstance(error, RuntimeError)
+ else type(error)
+ )
+ with pytest.raises(expected_error):
+ command.run()
+ persist.assert_called_once()
+ assert persist.call_args.args[4].execution_claimed
+
+
def test_match_slack_channel_rejects_ambiguous_casefolded_names() -> None:
channels: list[SlackChannel] = [
{"id": "C1", "name": "Private-Channel", "is_private": True},
@@ -3600,6 +4074,7 @@ def _make_state_instance(
schedule.retry_max_attempts = 3
schedule.retry_attempt = 0
schedule.retry_scheduled_dttm = None
+ schedule.execution_owner = None
schedule.send_failed_reports = False
schedule.retry_notify_owners = True
schedule.retry_notify_recipients = False
@@ -3613,6 +4088,9 @@ def test_working_state_timeout_raises_timeout_error(mocker: MockerFixture) -> No
"""Working state past timeout should raise WorkingTimeoutError and log ERROR."""
state = _make_state_instance(mocker, ReportWorkingState)
mocker.patch.object(state, "is_on_working_timeout", return_value=True)
+ mocker.patch(
+ "superset.commands.report.execute.db.session.query"
+ ).return_value.filter.return_value.update.return_value = 1
mock_log = mocker.Mock()
mock_log.end_dttm = datetime.utcnow() - timedelta(hours=2)
@@ -3660,6 +4138,9 @@ def test_working_timeout_replay_delegates_single_terminal_update(
last_state=ReportState.WORKING,
)
mocker.patch.object(state, "is_on_working_timeout", return_value=True)
+ mocker.patch(
+ "superset.commands.report.execute.db.session.query"
+ ).return_value.filter.return_value.update.return_value = 1
working_log = mocker.Mock()
working_log.uuid = state._execution_id
working_log.state = ReportState.WORKING
@@ -3691,6 +4172,9 @@ def test_stale_recovery_delegates_terminal_update_without_delivery(
last_state=ReportState.WORKING,
)
mocker.patch.object(state, "is_on_working_timeout", return_value=True)
+ mocker.patch(
+ "superset.commands.report.execute.db.session.query"
+ ).return_value.filter.return_value.update.return_value = 1
working_log = mocker.Mock()
working_log.uuid = uuid4()
working_log.state = ReportState.WORKING
@@ -4001,44 +4485,286 @@ def test_get_notification_content_alert_no_flag_skips_attachment(
assert content.text is None
-@pytest.mark.parametrize(
- ("schedule_type", "report_format", "attach_flag", "expected"),
- [
- # Reports always run under an execution context, regardless of format.
- (ReportScheduleType.REPORT, ReportDataFormat.PNG, False, True),
- (ReportScheduleType.REPORT, ReportDataFormat.PDF, False, True),
- (ReportScheduleType.REPORT, ReportDataFormat.CSV, False, True),
- (ReportScheduleType.REPORT, ReportDataFormat.TEXT, False, True),
- # Alerts that deliver a rendered screenshot fail closed only when the
- # ALERTS_ATTACH_REPORTS flag is on (otherwise no artifact is attached).
- (ReportScheduleType.ALERT, ReportDataFormat.PNG, True, True),
- (ReportScheduleType.ALERT, ReportDataFormat.PDF, True, True),
- (ReportScheduleType.ALERT, ReportDataFormat.PNG, False, False),
- (ReportScheduleType.ALERT, ReportDataFormat.PDF, False, False),
- # CSV/text/xlsx alerts never deliver a rendered screenshot; they stay
- # lenient even with the attach flag on.
- (ReportScheduleType.ALERT, ReportDataFormat.CSV, True, False),
- (ReportScheduleType.ALERT, ReportDataFormat.TEXT, True, False),
- (ReportScheduleType.ALERT, ReportDataFormat.XLSX, True, False),
- ],
-)
-@patch("superset.commands.report.execute.feature_flag_manager")
+@pytest.mark.parametrize("schedule_type", list(ReportScheduleType))
+@pytest.mark.parametrize("report_format", list(ReportDataFormat))
+@pytest.mark.parametrize("attach_flag", [False, True])
def test_should_build_execution_context(
- mock_ff: MagicMock,
mocker: MockerFixture,
schedule_type: ReportScheduleType,
report_format: ReportDataFormat,
attach_flag: bool,
- expected: bool,
) -> None:
- """Only reports and rendered-screenshot alerts run fail closed under a
- ReportExecutionContext; CSV/text alerts and flag-off alerts stay lenient."""
- mock_ff.is_feature_enabled.return_value = attach_flag
+ """Ownership fencing covers alerts even when they do not attach images."""
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=attach_flag,
+ )
model = mocker.Mock(spec=ReportSchedule)
model.type = schedule_type
model.report_format = report_format
+ assert _should_build_execution_context(model) is True
+ from superset.commands.report.execute import _uses_report_capture_contract
+
+ assert _uses_report_capture_contract(model) is (
+ schedule_type == ReportScheduleType.REPORT
+ or (
+ attach_flag
+ and report_format in (ReportDataFormat.PNG, ReportDataFormat.PDF)
+ )
+ )
- assert _should_build_execution_context(model) is expected
+
+@pytest.mark.parametrize("working_timeout", [7200, None])
+@pytest.mark.parametrize("kill", [False, True])
+def test_alert_deadline_preserves_schedule_timeout(
+ app: SupersetApp, working_timeout: int | None, kill: bool
+) -> None:
+ """Long-running alerts are not capped by the report-only global budget."""
+ from superset.commands.report.execute import _execution_budget_seconds
+
+ app.config["ALERT_REPORTS_EXECUTION_BUDGET_SECONDS"] = 3600
+ app.config["ALERT_REPORTS_WORKING_TIME_OUT_KILL"] = kill
+ app.config["ALERT_REPORTS_WORKING_SOFT_TIME_OUT_LAG"] = 10
+ model = ReportSchedule(
+ type=ReportScheduleType.ALERT, working_timeout=working_timeout
+ )
+ assert _execution_budget_seconds(model) == (
+ 7210 if kill and working_timeout else float("inf")
+ )
+
+
+@pytest.mark.parametrize("kill,working_timeout", [(False, 7200), (True, None)])
+@pytest.mark.parametrize(
+ "route", ["csv_get", "csv_post", "xlsx_get", "xlsx_post", "text"]
+)
+def test_unlimited_alert_data_transport(
+ app: SupersetApp,
+ mocker: MockerFixture,
+ kill: bool,
+ working_timeout: int | None,
+ route: str,
+) -> None:
+ """Exercise real urllib/socket requests for unlimited alert export settings."""
+ from dataclasses import replace
+ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+ from threading import Thread
+
+ from superset.commands.report.execute import _execution_budget_seconds
+
+ payload = json.dumps(
+ {
+ "result": [
+ {
+ "data": [{"value": 42}],
+ "coltypes": [0],
+ "colnames": ["value"],
+ "indexnames": [0],
+ }
+ ]
+ }
+ ).encode()
+ requests: list[str] = []
+
+ class Handler(BaseHTTPRequestHandler):
+ def do_GET(self) -> None: # noqa: N802
+ """Serve a valid embedded-data payload over a real socket."""
+ requests.append(self.command)
+ self.send_response(200)
+ self.send_header("Content-Length", str(len(payload)))
+ self.end_headers()
+ self.wfile.write(payload)
+
+ def do_POST(self) -> None: # noqa: N802
+ """Consume the export request before returning its bytes."""
+ self.rfile.read(int(self.headers.get("Content-Length", "0")))
+ self.do_GET()
+
+ def log_message(self, format: str, *args: Any) -> None:
+ """Keep test server access logs quiet."""
+
+ server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+ thread = Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+ try:
+ app.config["ALERT_REPORTS_WORKING_TIME_OUT_KILL"] = kill
+ app.config["ALERT_REPORTS_CSV_REQUEST_TIMEOUT"] = None
+ state = BaseReportState(
+ create_report_schedule(mocker), datetime.utcnow(), uuid4()
+ )
+ state._report_schedule.type = ReportScheduleType.ALERT
+ state._report_schedule.working_timeout = working_timeout
+ url_mock, _ = _mock_xlsx_chart_data_dependencies(mocker, state)
+ url = f"http://127.0.0.1:{server.server_port}/data"
+ url_mock.return_value = url
+ mocker.patch("superset.commands.report.execute.get_url_path", return_value=url)
+ mocker.patch.object(state, "_get_chart_data_request_payload", return_value={})
+ if route.endswith("post") or route == "text":
+ state._report_schedule.chart.query_context = "{}"
+ state._report_execution_context = replace(
+ _active_report_context(),
+ deadline=ReportExecutionDeadline(
+ total_seconds=_execution_budget_seconds(state._report_schedule)
+ ),
+ )
+ if route == "text":
+ assert state._get_embedded_data().iloc[0, 0] == 42
+ else:
+ result_format = (
+ ChartDataResultFormat.XLSX
+ if route.startswith("xlsx")
+ else ChartDataResultFormat.CSV
+ )
+ assert state._get_data(result_format) == payload
+ assert requests == ["POST" if route.endswith("post") else "GET"]
+ finally:
+ server.shutdown()
+ server.server_close()
+ thread.join(timeout=5)
+
+
+@pytest.mark.parametrize(
+ "budget,requested,expected",
+ [(float("inf"), None, None), (float("inf"), 5, 5), (10, None, 10), (10, 5, 5)],
+)
+def test_phase_timeout_preserves_finite_limits(
+ mocker: MockerFixture,
+ budget: float,
+ requested: float | None,
+ expected: float | None,
+) -> None:
+ """Only unbounded transport waits become None; finite caps stay enforced."""
+ from dataclasses import replace
+
+ state = _make_state_instance(mocker, BaseReportState)
+ state._report_execution_context = replace(
+ _active_report_context(),
+ deadline=ReportExecutionDeadline(
+ total_seconds=budget, started_at=0, _clock=lambda: 0
+ ),
+ )
+ assert (
+ state._phase_timeout("data_generation", requested_seconds=requested) == expected
+ )
+
+
+def test_send_error_restricts_diagnostics_to_editors(mocker: MockerFixture) -> None:
+ """Configured recipients must not receive the diagnostic editor-only notice."""
+ state = _make_state_instance(mocker, BaseReportState)
+ state._report_schedule.editors = _make_mock_editors(mocker, [1])
+ state._report_schedule.recipients = [mocker.Mock(spec=ReportRecipients)]
+ mocker.patch.object(
+ state, "_get_log_data", return_value=_make_notification_header()
+ )
+ mocker.patch.object(state, "_get_url", return_value="https://example.com")
+ send = mocker.patch.object(state, "_send")
+ state.send_error("failure", "useful diagnostic")
+ content, recipients = send.call_args.args
+ assert content.is_editor_error
+ assert content.text == "useful diagnostic"
+ assert [
+ json.loads(recipient.recipient_config_json)["target"]
+ for recipient in recipients
+ ] == ["user1@example.com"]
+
+
+@pytest.mark.parametrize(
+ "report_format",
+ [
+ ReportDataFormat.CSV,
+ ReportDataFormat.XLSX,
+ ReportDataFormat.TEXT,
+ ReportDataFormat.PNG,
+ ],
+)
+@pytest.mark.parametrize("unlimited", [False, True])
+def test_alert_bootstrap_retains_original_capture_contract(
+ mocker: MockerFixture, report_format: ReportDataFormat, unlimited: bool
+) -> None:
+ """Ownership context must not enable strict capture for data-only alerts."""
+ from dataclasses import replace
+ from math import isfinite
+
+ state = BaseReportState(create_report_schedule(mocker), datetime.utcnow(), uuid4())
+ state._report_schedule.type = ReportScheduleType.ALERT
+ state._report_schedule.report_format = report_format
+ state._report_execution_context = _active_report_context()
+ if unlimited:
+ state._report_execution_context = replace(
+ state._report_execution_context,
+ deadline=ReportExecutionDeadline(total_seconds=float("inf")),
+ )
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=True,
+ )
+ mocker.patch(
+ "superset.commands.report.execute.resolve_executor_user",
+ return_value=(mocker.Mock(), "executor"),
+ )
+ mocker.patch.object(state, "_get_url", return_value="https://example.com/chart")
+ screenshot = mocker.patch(
+ "superset.commands.report.execute.ChartScreenshot"
+ ).return_value.get_screenshot
+ screenshot.return_value = b"discarded capture"
+ assert state._get_screenshots() == [b"discarded capture"]
+ capture_context = screenshot.call_args.kwargs["report_execution_context"]
+ if report_format == ReportDataFormat.PNG:
+ assert capture_context is not None
+ assert isfinite(capture_context.deadline.total_seconds)
+ assert (
+ capture_context.deadline.total_seconds
+ <= state._report_execution_context.deadline.total_seconds
+ )
+ capture_context.reject_capture("test rejection")
+ assert state._report_execution_context.capture_was_rejected
+ else:
+ assert capture_context is None
+
+
+@pytest.mark.parametrize("global_enabled,opt_in", [(False, True), (True, False)])
+def test_disabled_retry_command_calls_fenced_cancellation(
+ mocker: MockerFixture, global_enabled: bool, opt_in: bool
+) -> None:
+ """The task boundary releases a matching disabled retry before returning."""
+ window = datetime(2026, 9, 15)
+ command = AsyncExecuteReportScheduleCommand(
+ str(uuid4()), 11, window, is_retry=True, expected_owner="owner"
+ )
+ command._model = ReportSchedule(
+ id=11,
+ retry_on_failure=opt_in,
+ last_state=ReportState.RETRYING,
+ retry_scheduled_dttm=window,
+ )
+ mocker.patch.object(command, "validate")
+ mocker.patch(
+ "superset.commands.report.execute.feature_flag_manager.is_feature_enabled",
+ return_value=global_enabled,
+ )
+ cancel = mocker.patch(
+ "superset.commands.report.execute.cancel_disabled_retry", return_value=True
+ )
+ machine = mocker.patch(
+ "superset.commands.report.execute.ReportScheduleStateMachine"
+ )
+ command.run()
+ cancel.assert_called_once_with(
+ mocker.ANY, 11, window, "owner", retries_enabled=global_enabled
+ )
+ machine.assert_not_called()
+
+
+def test_working_timeout_handles_missing_log_timestamp(mocker: MockerFixture) -> None:
+ """A legacy in-flight log may not have an end timestamp."""
+ state = _make_state_instance(mocker, BaseReportState)
+ state._report_schedule.execution_owner = None
+ state._report_schedule.last_eval_dttm = datetime.utcnow()
+ log = mocker.Mock(end_dttm=None)
+ mocker.patch(
+ "superset.commands.report.execute.ReportScheduleDAO.find_last_entered_working_log",
+ return_value=log,
+ )
+ assert state.is_on_working_timeout() is False
def test_create_log_success_commits(mocker: MockerFixture) -> None:
@@ -4207,7 +4933,11 @@ def test_terminal_persistence_retry_promotes_owned_working_execution(
working_log.error_message
== "Failed taking a screenshot readiness allocation expired"
)
- assert schedule.last_state == ReportState.ERROR
+ filtered_query.update.assert_called_once()
+ assert (
+ filtered_query.update.call_args.args[0][ReportSchedule.last_state]
+ == ReportState.ERROR
+ )
mock_db.session.commit.assert_called_once()
diff --git a/tests/unit_tests/commands/report/execution_claim_test.py b/tests/unit_tests/commands/report/execution_claim_test.py
new file mode 100644
index 00000000000..7baadc9314a
--- /dev/null
+++ b/tests/unit_tests/commands/report/execution_claim_test.py
@@ -0,0 +1,412 @@
+# 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 os
+from collections.abc import Iterator
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from threading import Barrier
+from typing import Any
+from unittest.mock import Mock
+from uuid import uuid4
+
+import pytest
+import sqlalchemy as sa
+from pytest_mock import MockerFixture
+from sqlalchemy.orm import sessionmaker
+
+from superset.commands.report.execution_claim import (
+ cancel_disabled_retry,
+ claim_execution,
+ normalize_window,
+)
+from superset.reports.models import ReportSchedule, ReportScheduleType, ReportState
+
+
+def utc_now() -> datetime:
+ """Match the metadata database's naive UTC timestamp representation."""
+ return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
+@pytest.fixture(params=["sqlite", "postgres"])
+def claim_engine(tmp_path: Path, request: pytest.FixtureRequest) -> Iterator[sa.Engine]:
+ """Exercise PostgreSQL REPEATABLE READ when a test database is provided."""
+ if request.param == "sqlite":
+ engine = sa.create_engine(f"sqlite:///{tmp_path / 'claims.db'}")
+ ReportSchedule.__table__.create(engine)
+ yield engine
+ engine.dispose()
+ return
+ uri = os.environ.get("REPORT_CLAIM_TEST_DATABASE_URI") or os.environ.get(
+ "SUPERSET__SQLALCHEMY_DATABASE_URI"
+ )
+ if not uri or sa.engine.make_url(uri).get_backend_name() != "postgresql":
+ pytest.skip("Set REPORT_CLAIM_TEST_DATABASE_URI for PostgreSQL race coverage")
+ schema = f"report_claim_test_{uuid4().hex}"
+ base_engine = sa.create_engine(uri, isolation_level="REPEATABLE READ")
+ table = ReportSchedule.__table__.to_metadata(sa.MetaData(), schema=schema)
+ with base_engine.begin() as connection:
+ connection.execute(sa.schema.CreateSchema(schema))
+ connection.execute(
+ sa.schema.CreateTable(table, include_foreign_key_constraints=[])
+ )
+ engine = base_engine.execution_options(schema_translate_map={None: schema})
+ try:
+ yield engine
+ finally:
+ with base_engine.begin() as connection:
+ connection.execute(sa.schema.DropSchema(schema, cascade=True))
+ base_engine.dispose()
+
+
+@pytest.fixture(params=[ReportScheduleType.REPORT, ReportScheduleType.ALERT])
+def sessions(claim_engine: sa.Engine, request: pytest.FixtureRequest) -> sessionmaker:
+ """Use real independent connections, not a shared mocked session."""
+ factory = sessionmaker(bind=claim_engine)
+ with factory() as session:
+ session.add(
+ ReportSchedule(
+ id=1,
+ name="claim test",
+ type=request.param,
+ crontab="* * * * *",
+ last_state=ReportState.NOOP,
+ retry_on_failure=True,
+ )
+ )
+ session.commit()
+ return factory
+
+
+def claim(session, window, *, retry=False, owner=None):
+ return claim_execution(
+ session,
+ 1,
+ str(uuid4()),
+ window,
+ is_retry=retry,
+ expected_owner=owner,
+ retries_enabled=True,
+ stale_retry_seconds=4140,
+ )
+
+
+@pytest.mark.parametrize("retry", [False, True])
+def test_competing_workers_have_one_winner(sessions, retry):
+ window = utc_now().replace(microsecond=0)
+ owner = str(uuid4())
+ if retry:
+ with sessions() as session:
+ session.query(ReportSchedule).update(
+ {
+ "last_state": ReportState.RETRYING,
+ "execution_owner": owner,
+ "execution_window": window,
+ "retry_scheduled_dttm": window,
+ }
+ )
+ session.commit()
+ # Synchronize AFTER both SELECTs. Each worker must contend on the UPDATE
+ # using the same observed state, rather than simply reading the winner.
+ barrier = Barrier(2)
+ engine = sessions.kw["bind"]
+
+ def synchronize(_conn, _cursor, statement, _parameters, _context, _executemany):
+ if statement.startswith("UPDATE ") and "report_schedule" in statement:
+ barrier.wait(timeout=10)
+
+ sa.event.listen(engine, "before_cursor_execute", synchronize)
+ try:
+
+ def run(_index):
+ with sessions() as session:
+ return claim(session, window, retry=retry, owner=owner)
+
+ with ThreadPoolExecutor(max_workers=2) as workers:
+ results = list(workers.map(run, range(2)))
+ assert sum(result is not None for result in results) == 1
+ finally:
+ sa.event.remove(engine, "before_cursor_execute", synchronize)
+
+
+def test_retry_window_and_owner_fence_replays(sessions):
+ window = utc_now().replace(microsecond=0)
+ with sessions() as session:
+ assert claim(session, window) is not None
+ schedule = session.query(ReportSchedule).one()
+ owner = schedule.execution_owner
+ schedule.last_state = ReportState.RETRYING
+ schedule.retry_scheduled_dttm = window
+ session.commit()
+ with sessions() as session:
+ assert claim(session, window) is None
+ assert claim(session, window, retry=True, owner="wrong-owner") is None
+ assert claim(session, window, retry=True, owner=owner) is not None
+ schedule = session.query(ReportSchedule).one()
+ schedule.last_state = ReportState.RETRYING
+ session.commit()
+ assert claim(session, window, retry=True, owner=owner) is None
+ schedule.last_state = ReportState.SUCCESS
+ session.commit()
+ assert claim(session, window) is None
+ assert claim(session, window + timedelta(hours=1)) is not None
+
+
+def test_alert_can_claim_a_retry(sessions):
+ window = utc_now()
+ with sessions() as session:
+ session.query(ReportSchedule).update(
+ {
+ "type": ReportScheduleType.ALERT,
+ "last_state": ReportState.RETRYING,
+ "execution_owner": "owner",
+ "retry_scheduled_dttm": window,
+ }
+ )
+ session.commit()
+ assert claim(session, window, retry=True, owner="owner") is not None
+
+
+def test_window_normalization_converts_offsets():
+ assert normalize_window(
+ datetime(2026, 9, 15, 10, tzinfo=timezone(timedelta(hours=10)))
+ ) == datetime(2026, 9, 15)
+
+
+def test_recovery_can_fence_worker_during_transport(
+ sessions: sessionmaker, mocker: MockerFixture
+) -> None:
+ """A separate connection can recover the row while a send is in flight."""
+ from superset.commands.report.exceptions import ReportSchedulePreviousWorkingError
+ from superset.commands.report.execute import BaseReportState
+ from superset.reports.notifications.base import NotificationContent
+ from superset.utils.report_execution import (
+ ReportExecutionContext,
+ ReportExecutionDeadline,
+ )
+
+ execution_id = uuid4()
+ with sessions() as session:
+ schedule = session.query(ReportSchedule).one()
+ schedule.execution_owner = str(execution_id)
+ schedule.last_state = ReportState.WORKING
+ session.commit()
+ context = ReportExecutionContext(
+ execution_id=execution_id,
+ report_schedule_id=1,
+ execution_claimed=True,
+ deadline=ReportExecutionDeadline(total_seconds=60),
+ )
+ state = BaseReportState(schedule, utc_now(), execution_id, context)
+ mocker.patch("superset.commands.report.execute.db.session", session)
+
+ def recover(*_args: Any) -> None:
+ with sessions() as recovery:
+ if recovery.bind.dialect.name == "postgresql":
+ recovery.execute(sa.text("SET LOCAL lock_timeout = '500ms'"))
+ assert (
+ recovery.query(ReportSchedule)
+ .filter_by(id=1)
+ .update({"execution_owner": None})
+ == 1
+ )
+ recovery.commit()
+
+ send = mocker.patch.object(state, "_send_notification", side_effect=recover)
+ with pytest.raises(ReportSchedulePreviousWorkingError):
+ state._send(
+ NotificationContent(
+ name="report",
+ text="failure",
+ header_data={
+ "notification_format": "PNG",
+ "editors": [],
+ "notification_type": "Report",
+ "notification_source": None,
+ "chart_id": None,
+ "dashboard_id": None,
+ "slack_channels": None,
+ "execution_id": str(execution_id),
+ },
+ ),
+ [Mock(), Mock()],
+ )
+ assert send.call_count == 1
+
+
+def test_terminal_fallback_persists_without_working_log(
+ sessions: sessionmaker, mocker: MockerFixture
+) -> None:
+ """A committed claim followed by rollback still gets a durable ERROR log."""
+ from superset.commands.report.execute import (
+ BaseReportState,
+ persist_owned_report_execution_terminal_error,
+ )
+ from superset.reports.models import ReportExecutionLog
+ from superset.utils.report_execution import (
+ ReportExecutionContext,
+ ReportExecutionDeadline,
+ )
+
+ engine = sessions.kw["bind"]
+ with engine.begin() as connection:
+ connection.execute(
+ sa.schema.CreateTable(
+ ReportExecutionLog.__table__, include_foreign_key_constraints=[]
+ )
+ )
+ execution_id = uuid4()
+ with sessions() as session:
+ assert claim_execution(
+ session,
+ 1,
+ str(execution_id),
+ utc_now(),
+ is_retry=False,
+ expected_owner=None,
+ retries_enabled=True,
+ stale_retry_seconds=4140,
+ )
+ mocker.patch("superset.commands.report.execute.db.session", session)
+ mocker.patch.object(
+ BaseReportState, "is_in_error_grace_period", return_value=False
+ )
+ send = mocker.patch.object(BaseReportState, "send_error")
+ context = ReportExecutionContext(
+ execution_id=execution_id,
+ report_schedule_id=1,
+ execution_claimed=True,
+ deadline=ReportExecutionDeadline(total_seconds=60),
+ )
+ assert persist_owned_report_execution_terminal_error(
+ 1, execution_id, "escaped", "RuntimeError", context
+ )
+ session.expire_all()
+ assert session.query(ReportSchedule).one().last_state == ReportState.ERROR
+ log = session.query(ReportExecutionLog).filter_by(error_message="escaped").one()
+ assert log.state == ReportState.ERROR
+ send.assert_called_once()
+ assert not persist_owned_report_execution_terminal_error(
+ 1, execution_id, "escaped", "RuntimeError", context
+ )
+ assert send.call_count == 1
+
+
+@pytest.mark.parametrize("sqlstate", ["40001", "40P01", "08006"])
+def test_claim_database_conflict_rolls_back(
+ mocker: MockerFixture, sqlstate: str
+) -> None:
+ """Only serialization/deadlock losers are suppressed, not connection failures."""
+ from sqlalchemy.exc import OperationalError
+
+ class DatabaseFailureError(Exception):
+ pgcode = sqlstate
+
+ original = DatabaseFailureError("database failure")
+ error = OperationalError("UPDATE", {}, original)
+ mocker.patch(
+ "superset.commands.report.execution_claim._claim_execution", side_effect=error
+ )
+ session = Mock()
+ if sqlstate == "08006":
+ with pytest.raises(OperationalError):
+ claim(session, utc_now())
+ else:
+ assert claim(session, utc_now()) is None
+ session.rollback.assert_called_once()
+
+
+@pytest.mark.parametrize("global_enabled,opt_in", [(False, True), (True, False)])
+@pytest.mark.parametrize("stale", [False, True])
+def test_cancel_disabled_retry_fences_owner_and_releases_next_window(
+ sessions: sessionmaker, global_enabled: bool, opt_in: bool, stale: bool
+) -> None:
+ """Disabling retries must not suspend cron or allow an old owner to cancel."""
+ window = utc_now().replace(microsecond=0)
+ with sessions() as session:
+ schedule = session.query(ReportSchedule).one()
+ schedule.last_state = ReportState.RETRYING
+ schedule.execution_owner = "owner"
+ schedule.execution_window = window
+ schedule.retry_scheduled_dttm = window
+ schedule.retry_on_failure = opt_in
+ schedule.retry_attempt = 2
+ session.commit()
+ assert cancel_disabled_retry(
+ session,
+ 1,
+ window,
+ "old-owner" if stale else "owner",
+ retries_enabled=global_enabled,
+ ) is (not stale)
+ session.expire_all()
+ assert schedule.last_state == (
+ ReportState.RETRYING if stale else ReportState.ERROR
+ )
+ if not stale:
+ assert schedule.retry_attempt == 0
+ assert schedule.retry_scheduled_dttm is None
+ assert claim(session, window) is None
+ assert claim(session, window + timedelta(minutes=5)) is not None
+
+
+def test_cancel_retry_cannot_cancel_new_window_or_reenabled_schedule(
+ sessions: sessionmaker,
+) -> None:
+ """Cancellation must recheck opt-in and the window in its UPDATE predicate."""
+ window = utc_now().replace(microsecond=0)
+ with sessions() as session:
+ schedule = session.query(ReportSchedule).one()
+ schedule.last_state = ReportState.RETRYING
+ schedule.execution_owner = "owner"
+ schedule.execution_window = window
+ schedule.retry_scheduled_dttm = window
+ session.commit()
+ assert not cancel_disabled_retry(
+ session, 1, window, "owner", retries_enabled=True
+ )
+ assert not cancel_disabled_retry(
+ session, 1, window - timedelta(minutes=5), "owner", retries_enabled=False
+ )
+
+
+@pytest.mark.parametrize("age,eligible", [(3599, False), (3601, True)])
+def test_stale_retry_recovers_at_max_delay(
+ sessions: sessionmaker, age: int, eligible: bool
+) -> None:
+ """Admission must not add the execution budget to the retry recovery delay."""
+ now = utc_now().replace(microsecond=0)
+ with sessions() as session:
+ schedule = session.query(ReportSchedule).one()
+ schedule.last_state = ReportState.RETRYING
+ schedule.execution_owner = "owner"
+ schedule.execution_window = now - timedelta(hours=2)
+ schedule.retry_scheduled_dttm = schedule.execution_window
+ schedule.last_eval_dttm = now - timedelta(seconds=age)
+ session.commit()
+ result = claim_execution(
+ session,
+ 1,
+ str(uuid4()),
+ now,
+ is_retry=False,
+ expected_owner=None,
+ retries_enabled=True,
+ stale_retry_seconds=3600,
+ )
+ assert (result is not None) is eligible
diff --git a/tests/unit_tests/commands/report/execution_ownership_migration_test.py b/tests/unit_tests/commands/report/execution_ownership_migration_test.py
new file mode 100644
index 00000000000..f078d2a7a62
--- /dev/null
+++ b/tests/unit_tests/commands/report/execution_ownership_migration_test.py
@@ -0,0 +1,87 @@
+# 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.
+
+"""Round-trip coverage for report execution ownership columns."""
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from alembic.script import ScriptDirectory
+
+
+def test_execution_ownership_migration_is_in_single_resolvable_chain() -> None:
+ """Resolve actual migration files so a phantom parent cannot pass CI."""
+ directory = Path(__file__).resolve().parents[4] / "superset/migrations"
+ scripts = ScriptDirectory(str(directory))
+ revisions = {revision.revision for revision in scripts.walk_revisions()}
+ assert "93d1b4a76c02" in revisions
+ assert len(scripts.get_heads()) == 1
+
+
+@pytest.mark.parametrize("partial_upgrade", [False, True])
+def test_execution_ownership_migration_round_trip(
+ partial_upgrade: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ """Preserve schedules and resume an interrupted add-column migration."""
+ path = Path(__file__).resolve().parents[4] / (
+ "superset/migrations/versions/"
+ "2026-09-16_00-00_93d1b4a76c02_add_report_execution_ownership.py"
+ )
+ spec = importlib.util.spec_from_file_location("execution_ownership_migration", path)
+ assert spec is not None
+ assert spec.loader is not None
+ migration = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(migration)
+ engine = sa.create_engine("sqlite://")
+ with engine.begin() as connection:
+ connection.exec_driver_sql(
+ "CREATE TABLE report_schedule (id INTEGER PRIMARY KEY, name VARCHAR(50))"
+ )
+ connection.exec_driver_sql(
+ "INSERT INTO report_schedule (id, name) VALUES (1, 'preserved')"
+ )
+ if partial_upgrade:
+ connection.exec_driver_sql(
+ "ALTER TABLE report_schedule ADD COLUMN execution_owner VARCHAR(36)"
+ )
+ monkeypatch.setattr(
+ migration, "op", Operations(MigrationContext.configure(connection))
+ )
+ migration.upgrade()
+ columns = {
+ c["name"]: c for c in sa.inspect(connection).get_columns("report_schedule")
+ }
+ assert columns["execution_owner"]["nullable"]
+ assert columns["execution_window"]["nullable"]
+ assert (
+ connection.exec_driver_sql("SELECT name FROM report_schedule").scalar()
+ == "preserved"
+ )
+ migration.downgrade()
+ assert {
+ c["name"] for c in sa.inspect(connection).get_columns("report_schedule")
+ } == {"id", "name"}
+ migration.upgrade()
+ assert (
+ connection.exec_driver_sql("SELECT COUNT(*) FROM report_schedule").scalar()
+ == 1
+ )
+ engine.dispose()
diff --git a/tests/unit_tests/commands/report/update_test.py b/tests/unit_tests/commands/report/update_test.py
index a78e65f6bef..8538f4bbfbe 100644
--- a/tests/unit_tests/commands/report/update_test.py
+++ b/tests/unit_tests/commands/report/update_test.py
@@ -516,17 +516,15 @@ def test_alert_with_nonexistent_database_rejected(mocker: MockerFixture) -> None
_PATCH_RETRY_FLAG = "superset.commands.report.update.is_feature_enabled"
-def test_update_rejects_retry_on_alert(mocker: MockerFixture) -> None:
- """Enabling retries on an alert schedule is rejected."""
+def test_update_accepts_retry_on_alert(mocker: MockerFixture) -> None:
+ """Alerts opt into the same bounded retry configuration as reports."""
model = _make_model(mocker, model_type=ReportScheduleType.ALERT, database_id=5)
_setup_mocks(mocker, model)
mocker.patch(_PATCH_RETRY_FLAG, return_value=True)
cmd = UpdateReportScheduleCommand(model_id=1, data={"retry_on_failure": True})
- with pytest.raises(ReportScheduleInvalidError) as exc_info:
- cmd.validate()
- messages = _get_validation_messages(exc_info)
- assert "retry_on_failure" in messages
+ cmd.validate()
+ assert cmd._properties["retry_on_failure"] is True
def test_update_rejects_send_failed_without_retry(mocker: MockerFixture) -> None:
diff --git a/tests/unit_tests/reports/notifications/email_tests.py b/tests/unit_tests/reports/notifications/email_tests.py
index fef59a3b4a7..dd8ca59b5f1 100644
--- a/tests/unit_tests/reports/notifications/email_tests.py
+++ b/tests/unit_tests/reports/notifications/email_tests.py
@@ -105,6 +105,41 @@ def test_error_template_sanitizes_html() -> None:
assert "<img" not in email_body
assert "<script>" not in email_body
assert "onerror=alert(1)" not in email_body
+ assert "DB error near" not in email_body
+ assert "Contact the report owner for error details." in email_body
+
+
+@pytest.mark.parametrize("editor_only", [False, True])
+def test_editor_error_diagnostics_are_sanitized(editor_only: bool) -> None:
+ """Only explicitly editor-restricted error emails preserve useful diagnostics."""
+ from superset.reports.models import ReportRecipients, ReportRecipientType
+ from superset.reports.notifications.base import NotificationContent
+ from superset.reports.notifications.email import EmailNotification
+
+ content = NotificationContent(
+ name="failure",
+ header_data={
+ "notification_format": "PNG",
+ "notification_type": "Alert",
+ "editors": [],
+ "notification_source": None,
+ "chart_id": None,
+ "dashboard_id": None,
+ "slack_channels": None,
+ "execution_id": "test",
+ },
+ is_editor_error=editor_only,
+ )
+ notification = EmailNotification(
+ ReportRecipients(type=ReportRecipientType.EMAIL), content
+ )
+ body = notification._error_template(
+ "DB query failed <script>alert(1)</script><img src=x onerror=alert(2)>"
+ )
+ assert ("DB query failed" in body) is editor_only
+ assert "<script" not in body
+ assert "<img" not in body
+ assert "onerror" not in body
def test_cta_link_included_by_default() -> None:
@@ -207,7 +242,8 @@ def test_error_template_cta_link_respects_include_cta() -> None:
._get_content()
.body
)
- assert "Report generation failed" in email_body
+ assert "Contact the report owner for error details." in email_body
+ assert "Report generation failed" not in email_body
assert "Explore in Superset" not in email_body
assert "http://example.com/superset/dashboard/1/" not in email_body
diff --git a/tests/unit_tests/reports/notifications/slack_tests.py b/tests/unit_tests/reports/notifications/slack_tests.py
index 7368c4606ac..7f1bb5d133c 100644
--- a/tests/unit_tests/reports/notifications/slack_tests.py
+++ b/tests/unit_tests/reports/notifications/slack_tests.py
@@ -63,6 +63,24 @@ if TYPE_CHECKING:
from superset.reports.notifications.slack import SlackNotification
+@pytest.mark.parametrize("attempt,maximum", [(None, None), (1, 3), (None, 3)])
+def test_error_notifications_redact_provider_diagnostics(
+ attempt: int | None, maximum: int | None
+) -> None:
+ """Retry and ordinary failures must not publish raw provider errors."""
+ from superset.reports.notifications.slack_mixin import SlackMixin
+
+ body = SlackMixin()._error_template(
+ "report",
+ "description",
+ "secret-provider-diagnostic",
+ retry_attempt=attempt,
+ retry_max_attempts=maximum,
+ )
+ assert "secret-provider-diagnostic" not in body
+ assert "Contact the report owner for error details." in body
+
+
@pytest.fixture(autouse=True)
def _skip_backoff_sleep():
"""Make phase-level Slack retry waits instant.
diff --git a/tests/unit_tests/reports/notifications/webhook_tests.py b/tests/unit_tests/reports/notifications/webhook_tests.py
index 49f2c75c8f3..e7efa083c3a 100644
--- a/tests/unit_tests/reports/notifications/webhook_tests.py
+++ b/tests/unit_tests/reports/notifications/webhook_tests.py
@@ -19,6 +19,7 @@
import pandas as pd
import pytest
from freezegun import freeze_time
+from pytest_mock import MockerFixture
from superset.reports.notifications.exceptions import (
NotificationParamException,
@@ -121,13 +122,81 @@ def test_get_req_payload_basic(mock_header_data) -> None:
assert payload["name"] == "Payload Name"
assert payload["description"] == "Payload Description"
assert payload["url"] == "http://example.com/report"
- assert payload["text"] == "Report Text"
+ assert payload["text"] == "Contact the report owner for error details."
assert isinstance(payload["header"], dict)
# Optional fields from header_data
assert payload["header"]["notification_format"] == "PNG"
assert payload["header"]["notification_type"] == "Alert"
+@pytest.mark.parametrize("notification_type", ["Alert", "Report"])
+@pytest.mark.parametrize("notice", ["ordinary", "retry", "final"])
+@pytest.mark.parametrize("multipart", [False, True])
+def test_send_redacts_failure_diagnostics(
+ mocker: MockerFixture,
+ mock_header_data: HeaderDataType,
+ notification_type: str,
+ notice: str,
+ multipart: bool,
+) -> None:
+ """Both webhook encodings redact all failure notices for alerts and reports."""
+ from superset.reports.models import ReportRecipients, ReportRecipientType
+ from superset.reports.notifications.base import NotificationContent
+
+ header = mock_header_data.copy()
+ header["notification_type"] = notification_type
+ raw_error = "Database password=private-value; SMTP recipient=private@example.com"
+ content = NotificationContent(
+ name="test report",
+ header_data=header,
+ text=raw_error,
+ retry_attempt=1 if notice == "retry" else None,
+ retry_max_attempts=3 if notice != "ordinary" else None,
+ csv=b"value\n42" if multipart else None,
+ )
+ notification = WebhookNotification(
+ ReportRecipients(
+ type=ReportRecipientType.WEBHOOK,
+ recipient_config_json='{"target": "https://example.com/webhook"}',
+ ),
+ content,
+ )
+ mocker.patch.object(notification, "_validate_webhook_url")
+ mocker.patch(
+ "superset.reports.notifications.webhook.feature_flag_manager.is_feature_enabled",
+ return_value=True,
+ )
+ requester = mocker.patch(
+ "superset.reports.notifications.webhook._get_requester"
+ ).return_value
+ requester.post.return_value.status_code = 200
+
+ notification.send()
+
+ requester.post.assert_called_once()
+ kwargs = requester.post.call_args.kwargs
+ payload = kwargs["data" if multipart else "json"]
+ assert payload["text"] == "Contact the report owner for error details."
+ assert "private-value" not in str(kwargs)
+ assert "private@example.com" not in str(kwargs)
+ assert content.text == raw_error
+
+
+@pytest.mark.parametrize("text", [None, ""])
+def test_success_payload_does_not_gain_error_text(
+ mock_header_data: HeaderDataType, text: str | None
+) -> None:
+ """Successful content retains the existing empty-text payload contract."""
+ from superset.reports.models import ReportRecipients, ReportRecipientType
+ from superset.reports.notifications.base import NotificationContent
+
+ notification = WebhookNotification(
+ ReportRecipients(type=ReportRecipientType.WEBHOOK),
+ NotificationContent(name="report", header_data=mock_header_data, text=text),
+ )
+ assert notification._get_req_payload()["text"] == text
+
+
def test_get_files_includes_all_content_types(mock_header_data) -> None:
"""
Test that _get_files correctly includes csv, pdf, and multiple screenshot attachments
diff --git a/tests/unit_tests/reports/schemas_test.py b/tests/unit_tests/reports/schemas_test.py
index 7f6e59fd78e..9d40debae3d 100644
--- a/tests/unit_tests/reports/schemas_test.py
+++ b/tests/unit_tests/reports/schemas_test.py
@@ -27,6 +27,34 @@ from superset.reports.schemas import (
)
+@pytest.mark.parametrize(
+ "schema_cls", [ReportSchedulePostSchema, ReportSchedulePutSchema]
+)
+@pytest.mark.parametrize("enabled", [True, False])
+def test_alert_retry_schema(
+ mocker: MockerFixture, schema_cls: type, enabled: bool
+) -> None:
+ """Both APIs accept alert retry settings only when the feature is enabled."""
+ mocker.patch("superset.reports.schemas.is_feature_enabled", return_value=enabled)
+ result = schema_cls().load(
+ {
+ "type": "Alert",
+ "name": "Retry alert",
+ "crontab": "* * * * *",
+ "database": 1,
+ "sql": "SELECT 1",
+ "validator_type": "not null",
+ "retry_on_failure": True,
+ "retry_max_attempts": 3,
+ }
+ )
+ if enabled:
+ assert result["retry_on_failure"] is True
+ assert result["retry_max_attempts"] == 3
+ else:
+ assert not result.get("retry_on_failure", False)
+
+
def test_report_post_schema_custom_width_validation(mocker: MockerFixture) -> None:
"""
Test the custom width validation.
diff --git a/tests/unit_tests/tasks/test_scheduler_soft_timeout.py b/tests/unit_tests/tasks/test_scheduler_soft_timeout.py
index 2c67dfb6764..8f686fde41a 100644
--- a/tests/unit_tests/tasks/test_scheduler_soft_timeout.py
+++ b/tests/unit_tests/tasks/test_scheduler_soft_timeout.py
@@ -22,6 +22,23 @@ import pytest
from celery.exceptions import SoftTimeLimitExceeded
+def test_legacy_retry_is_logged_without_executing() -> None:
+ """Legacy retry tasks are rejected observably rather than losing fencing."""
+ from superset.tasks.scheduler import execute
+
+ with (
+ patch("superset.tasks.scheduler.current_app") as app,
+ patch("superset.tasks.scheduler.AsyncExecuteReportScheduleCommand") as command,
+ patch("superset.tasks.scheduler.logger") as logger,
+ ):
+ execute(1234, "2026-09-15T00:00:00")
+ command.assert_not_called()
+ app.config["STATS_LOGGER"].incr.assert_any_call(
+ "reports.execute.legacy_retry_discarded"
+ )
+ assert "reason=missing_execution_owner" in logger.warning.call_args.args[0]
+
+
def test_soft_timeout_handler_is_shared_by_alerts() -> None:
"""The ``reports.execute`` soft-timeout handler is type-unconditional.