improve logic

superset-frontend/src/features/alerts/AlertReportModal.test.tsx

@@ -481,12 +481,12 @@ test('properly renders edit report text', async () => {
   expect(saveButton).toBeInTheDocument();
 });
 
-test('renders 4 sections for reports', () => {
+test('renders 5 sections for reports', () => {
   render(<AlertReportModal {...generateMockedProps(true)} />, {
     useRedux: true,
   });
   const sections = screen.getAllByRole('tab');
-  expect(sections.length).toBe(4);
+  expect(sections.length).toBe(5);
 });
 
 test('renders 5 sections for alerts', () => {
@@ -1475,7 +1475,7 @@ test('adding and removing dashboard filter rows', async () => {
 });
 
 test('alert shows condition section, report does not', () => {
-  // Alert has 5 sections
+  // Alert has 5 sections (general, condition, content, schedule, notification)
   const { unmount } = render(
     <AlertReportModal {...generateMockedProps(false)} />,
     { useRedux: true },
@@ -1484,11 +1484,11 @@ test('alert shows condition section, report does not', () => {
   expect(screen.getByTestId('alert-condition-panel')).toBeInTheDocument();
   unmount();
 
-  // Report has 4 sections, no condition panel
+  // Report has 5 sections (general, content, schedule, notification, error handling)
   render(<AlertReportModal {...generateMockedProps(true)} />, {
     useRedux: true,
   });
-  expect(screen.getAllByRole('tab')).toHaveLength(4);
+  expect(screen.getAllByRole('tab')).toHaveLength(5);
   expect(screen.queryByTestId('alert-condition-panel')).not.toBeInTheDocument();
 });
 
@@ -2879,3 +2879,40 @@ test('modal reopen resets local state', async () => {
     expect(screen.getByPlaceholderText(/enter report name/i)).toHaveValue('');
   });
 });
+
+// ---------- Error Handling Panel ----------
+
+test('renders error handling panel with Enable Retries switch', async () => {
+  render(<AlertReportModal {...generateMockedProps(true)} />, {
+    useRedux: true,
+  });
+  const errorHandlingTab = screen.getByText('Error handling');
+  expect(errorHandlingTab).toBeInTheDocument();
+  await userEvent.click(errorHandlingTab);
+  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);
+
+  // 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);
+
+  // 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();
+  });
+});

superset-frontend/src/features/alerts/AlertReportModal.tsx

@@ -722,6 +722,11 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
     validator_type: '',
     force_screenshot: false,
     grace_period: undefined,
+    retry_on_failure: false,
+    retry_max_attempts: 3,
+    send_failed_reports: false,
+    retry_notify_owners: true,
+    retry_notify_recipients: false,
   };
 
   const fetchDashboardFilterValues = async (
@@ -2730,6 +2735,115 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
                 </>
               ),
             },
+            ...(isReport
+              ? [
+                  {
+                    key: 'error-handling',
+                    label: (
+                      <CollapseLabelInModal
+                        title={t('Error handling')}
+                        subtitle={t(
+                          'Configure retry behavior on delivery failure.',
+                        )}
+                        testId="error-handling-panel"
+                      />
+                    ),
+                    children: (
+                      <div className="header-section">
+                        <StyledSwitchContainer
+                          css={css`
+                            margin-bottom: ${theme.sizeUnit * 4}px;
+                          `}
+                        >
+                          <Switch
+                            checked={!!currentAlert?.retry_on_failure}
+                            onChange={(checked: boolean) =>
+                              updateAlertState('retry_on_failure', checked)
+                            }
+                          />
+                          <div className="switch-label">
+                            {t('Enable Retries')}
+                          </div>
+                          <InfoTooltip
+                            tooltip={t(
+                              'Automatically retry sending the report when delivery fails.',
+                            )}
+                          />
+                        </StyledSwitchContainer>
+                        {currentAlert?.retry_on_failure && (
+                          <>
+                            <ModalFormField label={t('Maximum Retry Attempts')}>
+                              <InputNumber
+                                min={1}
+                                max={10}
+                                value={currentAlert?.retry_max_attempts ?? 3}
+                                onChange={(value: number | null) =>
+                                  updateAlertState(
+                                    'retry_max_attempts',
+                                    value ?? 3,
+                                  )
+                                }
+                              />
+                            </ModalFormField>
+                            <StyledSwitchContainer
+                              css={css`
+                                margin-bottom: ${theme.sizeUnit * 4}px;
+                              `}
+                            >
+                              <Switch
+                                checked={!!currentAlert?.send_failed_reports}
+                                onChange={(checked: boolean) =>
+                                  updateAlertState(
+                                    'send_failed_reports',
+                                    checked,
+                                  )
+                                }
+                              />
+                              <div className="switch-label">
+                                {t('Send Failed Reports')}
+                              </div>
+                              <InfoTooltip
+                                tooltip={t(
+                                  'By default, recipients only receive reports when all charts successfully load. ' +
+                                    'Enable this to send reports even when some charts fail to render.',
+                                )}
+                              />
+                            </StyledSwitchContainer>
+                            <ModalFormField label={t('Failure Notifications')}>
+                              <Checkbox
+                                checked={
+                                  currentAlert?.retry_notify_owners ?? true
+                                }
+                                onChange={(e: CheckboxChangeEvent) =>
+                                  updateAlertState(
+                                    'retry_notify_owners',
+                                    e.target.checked,
+                                  )
+                                }
+                              >
+                                {t('Owners')}
+                              </Checkbox>
+                              <Checkbox
+                                checked={
+                                  !!currentAlert?.retry_notify_recipients
+                                }
+                                onChange={(e: CheckboxChangeEvent) =>
+                                  updateAlertState(
+                                    'retry_notify_recipients',
+                                    e.target.checked,
+                                  )
+                                }
+                              >
+                                {t('Report Recipients')}
+                              </Checkbox>
+                            </ModalFormField>
+                          </>
+                        )}
+                      </div>
+                    ),
+                  },
+                ]
+              : []),
           ]}
         />
       </div>

superset-frontend/src/features/alerts/types.ts

@@ -151,6 +151,11 @@ export type AlertObject = {
   };
   validator_type?: string;
   working_timeout?: number;
+  retry_on_failure?: boolean;
+  retry_max_attempts?: number;
+  send_failed_reports?: boolean;
+  retry_notify_owners?: boolean;
+  retry_notify_recipients?: boolean;
 };
 
 export type LogObject = {

superset-frontend/src/features/reports/ReportModal/ReportModal.test.tsx

@@ -447,3 +447,80 @@ test('submit failure dispatches danger toast and keeps modal open', async () =>
 
   fetchMock.removeRoute('post-fail');
 });
+
+// ---------------------------------------------------------------------------
+// Error Handling section tests
+// ---------------------------------------------------------------------------
+
+test('Error Handling section is visible and Enable Retries checkbox is unchecked by default', () => {
+  const store = createStore({}, reducerIndex);
+  render(<ReportModal {...defaultProps} />, { useRedux: true, store });
+
+  expect(screen.getByText('Error Handling')).toBeInTheDocument();
+  const enableRetriesCheckbox = screen.getByRole('checkbox', {
+    name: /enable retries/i,
+  });
+  expect(enableRetriesCheckbox).not.toBeChecked();
+});
+
+test('conditional retry fields are hidden when Enable Retries is unchecked', () => {
+  const store = createStore({}, reducerIndex);
+  render(<ReportModal {...defaultProps} />, { useRedux: true, store });
+
+  expect(screen.queryByText('Maximum Retry Attempts')).not.toBeInTheDocument();
+  expect(screen.queryByText('Send Failed Reports')).not.toBeInTheDocument();
+  expect(screen.queryByText('Failure Notifications')).not.toBeInTheDocument();
+});
+
+test('conditional retry fields appear when Enable Retries is checked', async () => {
+  const store = createStore({}, reducerIndex);
+  render(<ReportModal {...defaultProps} />, { useRedux: true, store });
+
+  const enableRetriesCheckbox = screen.getByRole('checkbox', {
+    name: /enable retries/i,
+  });
+  await userEvent.click(enableRetriesCheckbox);
+
+  await waitFor(() => {
+    expect(screen.getByText('Maximum Retry Attempts')).toBeInTheDocument();
+    expect(screen.getByText('Send Failed Reports')).toBeInTheDocument();
+    expect(screen.getByText('Failure Notifications')).toBeInTheDocument();
+  });
+});
+
+test('retry fields are included in the POST body when Enable Retries is enabled', async () => {
+  fetchMock.post(REPORT_ENDPOINT, { result: {} }, { name: 'post-retry' });
+  const store = createStore({}, reducerIndex);
+  render(
+    <ReportModal
+      {...defaultProps}
+      dashboardId={undefined}
+      chart={{ sliceFormData: { viz_type: 'bar' } } as any}
+      creationMethod="charts"
+    />,
+    { useRedux: true, store },
+  );
+
+  // Enable retries
+  const enableRetriesCheckbox = screen.getByRole('checkbox', {
+    name: /enable retries/i,
+  });
+  await userEvent.click(enableRetriesCheckbox);
+
+  // Submit
+  const addButton = screen.getByRole('button', { name: /add/i });
+  await userEvent.click(addButton);
+
+  await waitFor(() => {
+    const calls = fetchMock.callHistory.calls('post-retry');
+    const lastCall = calls[calls.length - 1];
+    const body = JSON.parse(lastCall.options.body as string);
+    expect(body.retry_on_failure).toBe(true);
+    expect(typeof body.retry_max_attempts).toBe('number');
+    expect(typeof body.send_failed_reports).toBe('boolean');
+    expect(typeof body.retry_notify_owners).toBe('boolean');
+    expect(typeof body.retry_notify_recipients).toBe('boolean');
+  });
+
+  fetchMock.removeRoute('post-retry');
+});

superset-frontend/src/features/reports/ReportModal/index.tsx

@@ -35,10 +35,13 @@ import {
   subscribeReport,
 } from 'src/features/reports/ReportModal/actions';
 import {
+  Checkbox,
   Input,
   LabeledErrorBoundInput,
+  type CheckboxChangeEvent,
   type CronError,
 } from '@superset-ui/core/components';
+import { InputNumber } from '@superset-ui/core/components/Input';
 import TimezoneSelector from '@superset-ui/core/components/TimezoneSelector';
 import { Icons } from '@superset-ui/core/components/Icons';
 import { Typography } from '@superset-ui/core/components/Typography';
@@ -57,7 +60,9 @@ import { CreationMethod } from './HeaderReportDropdown';
 import {
   antDErrorAlertStyles,
   CustomWidthHeaderStyle,
+  StyledErrorHandlingSection,
   StyledModal,
+  StyledRetryFieldGroup,
   StyledTopSection,
   StyledBottomSection,
   StyledIconWrapper,
@@ -191,6 +196,11 @@ function ReportModal({
       crontab: currentReport.crontab,
       report_format: currentReport.report_format || defaultNotificationFormat,
       timezone: currentReport.timezone,
+      retry_on_failure: currentReport.retry_on_failure ?? false,
+      retry_max_attempts: currentReport.retry_max_attempts ?? 3,
+      send_failed_reports: currentReport.send_failed_reports ?? false,
+      retry_notify_owners: currentReport.retry_notify_owners ?? true,
+      retry_notify_recipients: currentReport.retry_notify_recipients ?? false,
     };
 
     setCurrentReport({ isSubmitting: true, error: undefined });
@@ -315,6 +325,75 @@ function ReportModal({
     </StyledInputContainer>
   );
 
+  const retryEnabled = !!currentReport.retry_on_failure;
+
+  const renderErrorHandlingSection = (
+    <StyledErrorHandlingSection>
+      <Typography.Title
+        level={4}
+        css={(theme: SupersetTheme) => SectionHeaderStyle(theme)}
+      >
+        {t('Error Handling')}
+      </Typography.Title>
+      <Checkbox
+        checked={retryEnabled}
+        onChange={(e: CheckboxChangeEvent) =>
+          setCurrentReport({ retry_on_failure: e.target.checked })
+        }
+      >
+        {t('Enable Retries')}
+      </Checkbox>
+      {retryEnabled && (
+        <StyledRetryFieldGroup>
+          <div>
+            <div className="control-label">{t('Maximum Retry Attempts')}</div>
+            <InputNumber
+              min={1}
+              max={10}
+              value={currentReport.retry_max_attempts ?? 3}
+              onChange={(value: number | null) =>
+                setCurrentReport({ retry_max_attempts: value ?? 3 })
+              }
+            />
+          </div>
+          <Checkbox
+            checked={!!currentReport.send_failed_reports}
+            onChange={(e: CheckboxChangeEvent) =>
+              setCurrentReport({ send_failed_reports: e.target.checked })
+            }
+          >
+            {t('Send Failed Reports')}
+          </Checkbox>
+          <div>
+            <div className="control-label">{t('Failure Notifications')}</div>
+            <div>
+              <Checkbox
+                checked={currentReport.retry_notify_owners ?? true}
+                onChange={(e: CheckboxChangeEvent) =>
+                  setCurrentReport({ retry_notify_owners: e.target.checked })
+                }
+              >
+                {t('Owners')}
+              </Checkbox>
+            </div>
+            <div>
+              <Checkbox
+                checked={!!currentReport.retry_notify_recipients}
+                onChange={(e: CheckboxChangeEvent) =>
+                  setCurrentReport({
+                    retry_notify_recipients: e.target.checked,
+                  })
+                }
+              >
+                {t('Report Recipients')}
+              </Checkbox>
+            </div>
+          </div>
+        </StyledRetryFieldGroup>
+      )}
+    </StyledErrorHandlingSection>
+  );
+
   return (
     <StyledModal
       show={show}
@@ -390,6 +469,7 @@ function ReportModal({
         />
         {isChart && renderMessageContentSection}
         {(!isChart || !isTextBasedChart) && renderCustomWidthSection}
+        {renderErrorHandlingSection}
       </StyledBottomSection>
       {currentReport.error && (
         <Alert

superset-frontend/src/features/reports/ReportModal/styles.tsx

@@ -113,3 +113,17 @@ export const antDErrorAlertStyles = (theme: SupersetTheme) => css`
   margin: ${theme.sizeUnit * 4}px;
   margin-top: 0;
 `;
+
+export const StyledErrorHandlingSection = styled.div`
+  margin-top: ${({ theme }) => theme.sizeUnit * 6}px;
+  border-top: 1px solid ${({ theme }) => theme.colorSplit};
+  padding-top: ${({ theme }) => theme.sizeUnit * 4}px;
+`;
+
+export const StyledRetryFieldGroup = styled.div`
+  margin-top: ${({ theme }) => theme.sizeUnit * 3}px;
+  padding-left: ${({ theme }) => theme.sizeUnit * 4}px;
+  display: flex;
+  flex-direction: column;
+  gap: ${({ theme }) => theme.sizeUnit * 3}px;
+`;

superset-frontend/src/features/reports/types.ts

@@ -66,4 +66,9 @@ export interface ReportObject {
   editors?: number[];
   custom_width?: number | null;
   error?: string;
+  retry_on_failure?: boolean;
+  retry_max_attempts?: number;
+  send_failed_reports?: boolean;
+  retry_notify_owners?: boolean;
+  retry_notify_recipients?: boolean;
 }

superset/commands/report/execute.py

@@ -1165,6 +1165,98 @@ class BaseReportState:
             < last_success.end_dttm
         )
 
+    def _get_retry_delay(self, attempt: int) -> int:
+        """Exponential backoff: base * 2^attempt, capped at a configurable max."""
+        base: int = app.config.get("ALERT_REPORTS_RETRY_BASE_DELAY_SECONDS", 60)
+        cap: int = app.config.get("ALERT_REPORTS_RETRY_MAX_DELAY_SECONDS", 3600)
+        return min(base * (2**attempt), cap)
+
+    def _is_retry_window_stale(self) -> bool:
+        """
+        Return True if a new crontab window fired while the previous one was
+        still retrying.  When stale the retry counter must be reset so the new
+        window gets a fresh retry budget.
+        """
+        anchor = self._report_schedule.retry_scheduled_dttm
+        return anchor is not None and anchor != self._scheduled_dttm
+
+    def _increment_retry(self) -> int:
+        """Increment retry_attempt, set the window anchor, and return the new count."""
+        self._report_schedule.retry_attempt += 1
+        self._report_schedule.retry_scheduled_dttm = self._scheduled_dttm
+        return self._report_schedule.retry_attempt
+
+    def _reset_retry_counter(self) -> None:
+        """Reset retry state after a terminal outcome."""
+        self._report_schedule.retry_attempt = 0
+        self._report_schedule.retry_scheduled_dttm = None
+
+    def _schedule_retry(self, delay_seconds: int) -> None:
+        """Re-queue the execute task with the given countdown (seconds)."""
+        # Lazy import to avoid a circular dependency between execute.py and scheduler.py
+        from superset.tasks.scheduler import execute as execute_task  # noqa: PLC0415
+
+        execute_task.apply_async(
+            (self._report_schedule.id,),
+            countdown=delay_seconds,
+        )
+
+    def send_retry_notification(
+        self, attempt: int, max_attempts: int, error_message: str
+    ) -> None:
+        """
+        Send a per-retry-attempt failure notification to the owners and/or recipients
+        selected via the retry_notify_owners / retry_notify_recipients flags.
+        """
+        recipients: list[ReportRecipients] = []
+
+        if self._report_schedule.retry_notify_owners:
+            recipients.extend(
+                [
+                    ReportRecipients(
+                        type=ReportRecipientType.EMAIL,
+                        recipient_config_json=json.dumps({"target": s.user.email}),
+                    )
+                    for s in self._report_schedule.editors
+                    if s.type == SubjectType.USER and s.user
+                ]
+            )
+
+        if self._report_schedule.retry_notify_recipients:
+            recipients.extend(self._report_schedule.recipients)
+
+        if not recipients:
+            return
+
+        header_data = self._get_log_data()
+        url = self._get_url(user_friendly=True)
+        notification_content = NotificationContent(
+            name=sanitize_title(self._report_schedule.name),
+            text=error_message,
+            header_data=header_data,
+            url=url,
+            retry_attempt=attempt,
+            retry_max_attempts=max_attempts,
+        )
+        self._send(notification_content, recipients)
+
+    def send_final_failure_report(self, error_message: str) -> None:
+        """
+        Send the failed report notification to all configured recipients after
+        all retry attempts have been exhausted and send_failed_reports is enabled.
+        """
+        header_data = self._get_log_data()
+        url = self._get_url(user_friendly=True)
+        max_attempts: int = self._report_schedule.retry_max_attempts
+        notification_content = NotificationContent(
+            name=sanitize_title(self._report_schedule.name),
+            text=error_message,
+            header_data=header_data,
+            url=url,
+            retry_max_attempts=max_attempts,
+        )
+        self._send(notification_content, self._report_schedule.recipients)
+
     def is_on_working_timeout(self) -> bool:
         """
         Checks if an alert is in a working timeout
@@ -1195,7 +1287,7 @@ class ReportNotTriggeredErrorState(BaseReportState):
     - Error
     """
 
-    current_states = [ReportState.NOOP, ReportState.ERROR]
+    current_states = [ReportState.NOOP, ReportState.ERROR, ReportState.RETRYING]
     initial = True
 
     def next(self) -> None:  # noqa: C901
@@ -1224,6 +1316,60 @@ class ReportNotTriggeredErrorState(BaseReportState):
             if isinstance(first_ex, SupersetErrorsException):
                 error_message = ";".join([error.message for error in first_ex.errors])
 
+            # --- Retry logic ---
+            retry_on_failure: bool = self._report_schedule.retry_on_failure
+            max_attempts: int = self._report_schedule.retry_max_attempts
+
+            # If a new crontab window has fired since the first failure, reset the
+            # retry counter so this window gets a fresh budget.
+            if retry_on_failure and self._is_retry_window_stale():
+                self._reset_retry_counter()
+
+            current_attempt = self._report_schedule.retry_attempt
+
+            if retry_on_failure and current_attempt < max_attempts:
+                # Schedule another attempt and exit cleanly (don't re-raise).
+                attempt = self._increment_retry()
+                try:
+                    self.update_report_schedule_and_log(
+                        ReportState.RETRYING, error_message=error_message
+                    )
+                except ReportScheduleUnexpectedError as logging_ex:
+                    logger.warning(
+                        "Failed to log RETRYING state for report schedule "
+                        "(execution %s) due to database issue",
+                        self._execution_id,
+                        exc_info=True,
+                    )
+                    raise first_ex from logging_ex
+                try:
+                    self.send_retry_notification(attempt, max_attempts, error_message)
+                except Exception:  # pylint: disable=broad-except
+                    logger.warning(
+                        "Failed to send retry notification for report schedule "
+                        "(execution %s)",
+                        self._execution_id,
+                        exc_info=True,
+                    )
+                self._schedule_retry(self._get_retry_delay(attempt))
+                return  # task completes normally; next attempt is queued
+
+            # All retries exhausted (or retry disabled) — fall through to the
+            # existing error-logging and grace-period notification path.
+            if retry_on_failure:
+                # Reset counter so the next crontab window starts fresh.
+                self._reset_retry_counter()
+                if self._report_schedule.send_failed_reports:
+                    try:
+                        self.send_final_failure_report(error_message)
+                    except Exception:  # pylint: disable=broad-except
+                        logger.warning(
+                            "Failed to send final failure report for report schedule "
+                            "(execution %s)",
+                            self._execution_id,
+                            exc_info=True,
+                        )
+
             try:
                 self.update_report_schedule_and_log(
                     ReportState.ERROR, error_message=error_message
@@ -1403,6 +1549,8 @@ class ReportSuccessState(BaseReportState):
             warning_message = (
                 ";".join(self._filter_warnings) if self._filter_warnings else None
             )
+            # Clear any retry state from previous failed attempts in this window.
+            self._reset_retry_counter()
             self.update_report_schedule_and_log(
                 ReportState.SUCCESS, error_message=warning_message
             )

superset/migrations/versions/2026-07-24_00-00_f3a8c1d2e9b7_add_report_retry_state_columns.py

@@ -0,0 +1,84 @@
+# 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_retry_columns
+
+Revision ID: f3a8c1d2e9b7
+Revises: e5f6a7b8c9d0
+Create Date: 2026-07-24 00:00:00.000000
+
+"""
+
+import logging
+
+import sqlalchemy as sa
+from alembic import op
+
+from superset.migrations.shared.utils import get_table_column
+
+logger = logging.getLogger("alembic.env")
+
+# revision identifiers, used by Alembic.
+revision = "f3a8c1d2e9b7"
+down_revision = "e5f6a7b8c9d0"
+
+# Configuration columns (user-configurable)
+_CONFIG_COLUMNS = [
+    ("retry_on_failure", sa.Boolean(), False, "0"),
+    ("retry_max_attempts", sa.Integer(), False, "3"),
+    ("send_failed_reports", sa.Boolean(), False, "0"),
+    ("retry_notify_owners", sa.Boolean(), False, "1"),
+    ("retry_notify_recipients", sa.Boolean(), False, "0"),
+]
+
+# State columns (written by the execution engine)
+_STATE_COLUMNS = [
+    ("retry_attempt", sa.Integer(), False, "0"),
+    ("retry_scheduled_dttm", sa.DateTime(), True, None),
+]
+
+
+def upgrade() -> None:
+    """Add retry config and state columns to report_schedule."""
+    if get_table_column("report_schedule", "retry_attempt") is not None:
+        logger.info(
+            "Column report_schedule.retry_attempt already exists. Skipping migration."
+        )
+        return
+
+    with op.batch_alter_table("report_schedule") as batch_op:
+        for name, col_type, nullable, default in _CONFIG_COLUMNS + _STATE_COLUMNS:
+            batch_op.add_column(
+                sa.Column(
+                    name,
+                    col_type,
+                    nullable=nullable,
+                    server_default=default,
+                )
+            )
+
+
+def downgrade() -> None:
+    """Remove retry config and state columns from report_schedule."""
+    if get_table_column("report_schedule", "retry_attempt") is None:
+        logger.info(
+            "Column report_schedule.retry_attempt does not exist. Skipping downgrade."
+        )
+        return
+
+    with op.batch_alter_table("report_schedule") as batch_op:
+        for name, _, _, _ in _CONFIG_COLUMNS + _STATE_COLUMNS:
+            batch_op.drop_column(name)

superset/reports/models.py

@@ -77,6 +77,7 @@ class ReportState(StrEnum):
     ERROR = "Error"
     NOOP = "Not triggered"
     GRACE = "On Grace"
+    RETRYING = "Retrying"
 
 
 class ReportDataFormat(StrEnum):
@@ -164,6 +165,25 @@ class ReportSchedule(AuditMixinNullable, ExtraJSONMixin, Model):
     custom_width = Column(Integer, nullable=True)
     custom_height = Column(Integer, nullable=True)
 
+    # Retry configuration — user-configurable
+    retry_on_failure = Column(
+        Boolean, default=False, nullable=False, server_default="0"
+    )
+    retry_max_attempts = Column(Integer, default=3, nullable=False, server_default="3")
+    send_failed_reports = Column(
+        Boolean, default=False, nullable=False, server_default="0"
+    )
+    retry_notify_owners = Column(
+        Boolean, default=True, nullable=False, server_default="1"
+    )
+    retry_notify_recipients = Column(
+        Boolean, default=False, nullable=False, server_default="0"
+    )
+
+    # 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)
+
     extra: ReportScheduleExtra  # type: ignore
 
     email_subject = Column(String(255))

superset/reports/notifications/base.py

@@ -35,6 +35,9 @@ class NotificationContent:
     description: Optional[str] = ""
     url: Optional[str] = None  # url to chart/dashboard for this screenshot
     embedded_data: Optional[pd.DataFrame] = None
+    # Populated only when this is a per-retry or final-failure notification
+    retry_attempt: Optional[int] = None
+    retry_max_attempts: Optional[int] = None
 
 
 class BaseNotification:  # pylint: disable=too-few-public-methods

superset/reports/notifications/email.py

@@ -161,7 +161,71 @@ class EmailNotification(BaseNotification):  # pylint: disable=too-few-public-met
             call_to_action=call_to_action,
         )
 
+    def _retry_error_template(self, text: str) -> tuple[str, dict[str, bytes]]:
+        """HTML body for a per-retry-attempt failure notification."""
+        attempt = self._content.retry_attempt
+        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={})
+        call_to_action = self._get_call_to_action()
+
+        img_tags = ""
+        if self._content.screenshots:
+            domain = self._get_smtp_domain()
+            images = {
+                make_msgid(domain)[1:-1]: screenshot
+                for screenshot in self._content.screenshots
+            }
+            img_parts = [
+                f'<div class="image"><img width="1000" src="cid:{msgid}"></div>'
+                for msgid in images
+            ]
+            img_tags = "".join(img_parts)
+        else:
+            images = {}
+
+        body = textwrap.dedent(
+            f"""
+            <html>
+              <head>
+                <style type="text/css">
+                  table, th, td {{
+                    border-collapse: collapse;
+                    border-color: rgb(200, 212, 227);
+                    color: rgb(42, 63, 95);
+                    padding: 4px 8px;
+                  }}
+                  .image {{ margin-bottom: 18px; min-width: 1000px; }}
+                </style>
+              </head>
+              <body>
+                <h3>Report Generation Failed &mdash; Retry in Progress</h3>
+                <p>
+                  Your scheduled report
+                  <b>{nh3.clean(self._content.name, tags=set(), attributes={})}</b>
+                  encountered an error during generation.
+                  The system is automatically retrying.
+                </p>
+                <p><b>Retry attempt:</b> {attempt} of {max_attempts}
+                   &nbsp;&nbsp; <b>Retries remaining:</b> {retries_remaining}</p>
+                <p><b>Error details:</b> {safe_text}</p>
+                <p><b><a href="{self._content.url}">{call_to_action}</a></b></p>
+                {img_tags}
+              </body>
+            </html>
+            """
+        )
+        return body, images
+
     def _get_content(self) -> EmailContent:
+        if self._content.text and self._content.retry_attempt is not None:
+            body, images = self._retry_error_template(self._content.text)
+            return EmailContent(
+                body=body,
+                images=images or None,
+                header_data=self._content.header_data,
+            )
         if self._content.text:
             return EmailContent(body=self._error_template(self._content.text))
         # Get the domain from the 'From' address ..
@@ -267,6 +331,19 @@ class EmailNotification(BaseNotification):  # pylint: disable=too-few-public-met
         )
 
     def _get_subject(self) -> str:
+        if self._content.retry_attempt is not None:
+            return __(
+                "Report Retry [%(attempt)s of %(max)s]: %(name)s",
+                attempt=self._content.retry_attempt,
+                max=self._content.retry_max_attempts,
+                name=self._name,
+            )
+        if self._content.text and self._content.retry_max_attempts is not None:
+            # Final-failure notification (retry_attempt is None but max is set)
+            return __(
+                "Report Failed - All Retries Exhausted: %(name)s",
+                name=self._name,
+            )
         return __(
             "%(prefix)s %(title)s",
             prefix=current_app.config["EMAIL_REPORTS_SUBJECT_PREFIX"],

superset/reports/notifications/slack_mixin.py

@@ -47,7 +47,31 @@ class SlackMixin:
         )
 
     @staticmethod
-    def _error_template(name: str, description: str, text: str) -> str:
+    def _error_template(
+        name: str,
+        description: str,
+        text: str,
+        retry_attempt: int | None = None,
+        retry_max_attempts: int | None = None,
+    ) -> str:
+        if retry_attempt is not None:
+            retries_remaining = (retry_max_attempts or 0) - retry_attempt
+            return __(
+                """*Report Retry [%(attempt)s of %(max)s]: %(name)s*
+
+%(description)s
+
+Retry attempt: %(attempt)s of %(max)s  |  Retries remaining: %(remaining)s
+
+Error: %(text)s
+""",
+                name=name,
+                description=description,
+                attempt=retry_attempt,
+                max=retry_max_attempts,
+                remaining=retries_remaining,
+                text=text,
+            )
         return __(
             """*%(name)s*
 
@@ -63,7 +87,11 @@ class SlackMixin:
     def _get_body(self, content: NotificationContent) -> str:
         if content.text:
             return self._error_template(
-                content.name, content.description or "", content.text
+                content.name,
+                content.description or "",
+                content.text,
+                retry_attempt=content.retry_attempt,
+                retry_max_attempts=content.retry_max_attempts,
             )
 
         if content.embedded_data is None:

superset/reports/schemas.py

@@ -278,6 +278,35 @@ class ReportSchedulePostSchema(Schema):
         required=False,
         dump_default=None,
     )
+    retry_on_failure = fields.Boolean(
+        metadata={"description": _("Enable automatic retries on report failure")},
+        load_default=False,
+    )
+    retry_max_attempts = fields.Integer(
+        metadata={
+            "description": _("Maximum number of retry attempts (1–10)"),
+            "example": 3,
+        },
+        load_default=3,
+        required=False,
+        validate=[Range(min=1, max=10, error=_("Must be between 1 and 10"))],
+    )
+    send_failed_reports = fields.Boolean(
+        metadata={
+            "description": _(
+                "Send the failed report to all recipients after retries are exhausted"
+            )
+        },
+        load_default=False,
+    )
+    retry_notify_owners = fields.Boolean(
+        metadata={"description": _("Notify report owners on each retry attempt")},
+        load_default=True,
+    )
+    retry_notify_recipients = fields.Boolean(
+        metadata={"description": _("Notify report recipients on each retry attempt")},
+        load_default=False,
+    )
 
     @validates("custom_width")
     def validate_custom_width(
@@ -311,6 +340,21 @@ class ReportSchedulePostSchema(Schema):
                     {"database": ["Database reference is not allowed on a report"]}
                 )
 
+    @validates_schema
+    def validate_retry_config(  # pylint: disable=unused-argument
+        self,
+        data: dict[str, Any],
+        **kwargs: Any,
+    ) -> None:
+        if data.get("send_failed_reports") and not data.get("retry_on_failure"):
+            raise ValidationError(
+                {
+                    "send_failed_reports": [
+                        _("send_failed_reports requires retry_on_failure to be enabled")
+                    ]
+                }
+            )
+
 
 class ReportScheduleSubscribeSchema(ReportSchedulePostSchema):
     """Schema for creating a chart/dashboard subscription.
@@ -441,6 +485,34 @@ class ReportSchedulePutSchema(Schema):
         required=False,
         dump_default=None,
     )
+    retry_on_failure = fields.Boolean(
+        metadata={"description": _("Enable automatic retries on report failure")},
+        required=False,
+    )
+    retry_max_attempts = fields.Integer(
+        metadata={
+            "description": _("Maximum number of retry attempts (1–10)"),
+            "example": 3,
+        },
+        required=False,
+        validate=[Range(min=1, max=10, error=_("Must be between 1 and 10"))],
+    )
+    send_failed_reports = fields.Boolean(
+        metadata={
+            "description": _(
+                "Send the failed report to all recipients after retries are exhausted"
+            )
+        },
+        required=False,
+    )
+    retry_notify_owners = fields.Boolean(
+        metadata={"description": _("Notify report owners on each retry attempt")},
+        required=False,
+    )
+    retry_notify_recipients = fields.Boolean(
+        metadata={"description": _("Notify report recipients on each retry attempt")},
+        required=False,
+    )
 
     @validates("custom_width")
     def validate_custom_width(
@@ -462,6 +534,21 @@ class ReportSchedulePutSchema(Schema):
                 )
             )
 
+    @validates_schema
+    def validate_retry_config(  # pylint: disable=unused-argument
+        self,
+        data: dict[str, Any],
+        **kwargs: Any,
+    ) -> None:
+        if data.get("send_failed_reports") and not data.get("retry_on_failure"):
+            raise ValidationError(
+                {
+                    "send_failed_reports": [
+                        _("send_failed_reports requires retry_on_failure to be enabled")
+                    ]
+                }
+            )
+
 
 class SlackChannelSchema(Schema):
     """

superset/reports/types.py

@@ -19,5 +19,10 @@ from typing import TypedDict
 from superset.dashboards.permalink.types import DashboardPermalinkState
 
 
-class ReportScheduleExtra(TypedDict):
+class ReportScheduleExtra(TypedDict, total=False):
     dashboard: DashboardPermalinkState
+    retry_on_failure: bool
+    retry_max_attempts: int
+    send_failed_reports: bool
+    retry_notify_owners: bool
+    retry_notify_recipients: bool

tests/integration_tests/reports/commands_tests.py

@@ -2667,3 +2667,178 @@ def test__send_with_server_errors(notification_mock, logger_mock):
     logger_mock.warning.assert_called_with(
         "SupersetError(message='', error_type=<SupersetErrorType.REPORT_NOTIFICATION_ERROR: 'REPORT_NOTIFICATION_ERROR'>, level=<ErrorLevel.ERROR: 'error'>, extra=None)"  # noqa: E501
     )
+
+
+# ---------------------------------------------------------------------------
+# Retry tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
+@patch("superset.commands.report.execute.ReportNotTriggeredErrorState._schedule_retry")
+@patch("superset.reports.notifications.email.send_email_smtp")
+@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
+def test_retry_on_failure_schedules_retry(
+    screenshot_mock: Mock,
+    email_mock: Mock,
+    schedule_retry_mock: Mock,
+) -> None:
+    """
+    ExecuteReport Command: when retry_on_failure is enabled and the report fails,
+    the state transitions to RETRYING and a retry task is enqueued.
+    """
+    chart = db.session.query(Slice).first()
+    report_schedule = create_report_notification(
+        email_target="target@email.com",
+        chart=chart,
+        retry_on_failure=True,
+        retry_max_attempts=3,
+        retry_notify_owners=False,
+        retry_notify_recipients=False,
+    )
+    try:
+        screenshot_mock.side_effect = Exception("screenshot failed")
+
+        # Should NOT re-raise (retry path exits cleanly)
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, report_schedule.id, datetime.utcnow()
+        ).run()
+
+        db.session.refresh(report_schedule)
+        assert report_schedule.last_state == ReportState.RETRYING
+        assert report_schedule.retry_attempt == 1
+        schedule_retry_mock.assert_called_once()
+        # No error email should be sent on the retry path
+        email_mock.assert_not_called()
+    finally:
+        cleanup_report_schedule(report_schedule)
+
+
+@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
+@patch("superset.commands.report.execute.ReportNotTriggeredErrorState._schedule_retry")
+@patch("superset.reports.notifications.email.send_email_smtp")
+@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
+def test_retry_exhausted_transitions_to_error(
+    screenshot_mock: Mock,
+    email_mock: Mock,
+    schedule_retry_mock: Mock,
+) -> None:
+    """
+    ExecuteReport Command: when all retries are exhausted the state transitions
+    to ERROR, the retry counter is reset, and an error email is sent to editors.
+    """
+    chart = db.session.query(Slice).first()
+    report_schedule = create_report_notification(
+        email_target="target@email.com",
+        chart=chart,
+        retry_on_failure=True,
+        retry_max_attempts=2,
+        retry_notify_owners=False,
+        retry_notify_recipients=False,
+    )
+    # Pre-set retry_attempt to the max so the next execution exhausts retries.
+    report_schedule.retry_attempt = 2
+    report_schedule.retry_scheduled_dttm = datetime.utcnow()
+    db.session.commit()
+
+    try:
+        screenshot_mock.side_effect = Exception("screenshot failed")
+
+        with pytest.raises(Exception, match="screenshot failed"):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, report_schedule.id, datetime.utcnow()
+            ).run()
+
+        db.session.refresh(report_schedule)
+        assert report_schedule.last_state == ReportState.ERROR
+        # Counter is reset after exhaustion
+        assert report_schedule.retry_attempt == 0
+        # No further retry should have been scheduled
+        schedule_retry_mock.assert_not_called()
+        # Error email sent to editors
+        assert email_mock.call_count >= 1
+    finally:
+        cleanup_report_schedule(report_schedule)
+
+
+@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
+@patch("superset.commands.report.execute.ReportNotTriggeredErrorState._schedule_retry")
+@patch("superset.commands.report.execute.BaseReportState._send")
+@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
+def test_send_failed_reports_sends_to_recipients(
+    screenshot_mock: Mock,
+    send_mock: Mock,
+    schedule_retry_mock: Mock,
+) -> None:
+    """
+    ExecuteReport Command: when send_failed_reports is True and all retries are
+    exhausted, _send is called with the configured recipients.
+    """
+    chart = db.session.query(Slice).first()
+    report_schedule = create_report_notification(
+        email_target="target@email.com",
+        chart=chart,
+        retry_on_failure=True,
+        retry_max_attempts=1,
+        send_failed_reports=True,
+        retry_notify_owners=False,
+        retry_notify_recipients=False,
+    )
+    report_schedule.retry_attempt = 1
+    report_schedule.retry_scheduled_dttm = datetime.utcnow()
+    db.session.commit()
+
+    try:
+        screenshot_mock.side_effect = Exception("screenshot failed")
+
+        with pytest.raises(Exception, match="screenshot failed"):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, report_schedule.id, datetime.utcnow()
+            ).run()
+
+        # _send should have been called for the final failure report
+        assert send_mock.call_count >= 1
+        schedule_retry_mock.assert_not_called()
+    finally:
+        cleanup_report_schedule(report_schedule)
+
+
+@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
+@patch("superset.commands.report.execute.ReportNotTriggeredErrorState._schedule_retry")
+@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
+def test_retrying_state_routes_back_to_error_handler(
+    screenshot_mock: Mock,
+    schedule_retry_mock: Mock,
+) -> None:
+    """
+    ExecuteReport Command: a schedule with last_state=RETRYING is routed to
+    ReportNotTriggeredErrorState, which can retry again or exhaust and error.
+    """
+    chart = db.session.query(Slice).first()
+    report_schedule = create_report_notification(
+        email_target="target@email.com",
+        chart=chart,
+        retry_on_failure=True,
+        retry_max_attempts=3,
+        retry_notify_owners=False,
+        retry_notify_recipients=False,
+    )
+    report_schedule.last_state = ReportState.RETRYING
+    report_schedule.retry_attempt = 1
+    report_schedule.retry_scheduled_dttm = datetime.utcnow()
+    db.session.commit()
+
+    try:
+        screenshot_mock.side_effect = Exception("still failing")
+
+        # Should not raise — still within retry budget
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, report_schedule.id, datetime.utcnow()
+        ).run()
+
+        db.session.refresh(report_schedule)
+        assert report_schedule.last_state == ReportState.RETRYING
+        assert report_schedule.retry_attempt == 2
+        schedule_retry_mock.assert_called_once()
+    finally:
+        cleanup_report_schedule(report_schedule)

tests/integration_tests/reports/utils.py

@@ -85,6 +85,11 @@ def insert_report_schedule(
     logs: Optional[list[ReportExecutionLog]] = None,
     extra: Optional[dict[Any, Any]] = None,
     force_screenshot: bool = False,
+    retry_on_failure: bool = False,
+    retry_max_attempts: int = 3,
+    send_failed_reports: bool = False,
+    retry_notify_owners: bool = True,
+    retry_notify_recipients: bool = False,
 ) -> ReportSchedule:
     editors = editors or []
     editor_users = [s.user for s in editors if s.type == SubjectType.USER and s.user]
@@ -114,6 +119,11 @@ def insert_report_schedule(
             report_format=report_format,
             extra=extra,
             force_screenshot=force_screenshot,
+            retry_on_failure=retry_on_failure,
+            retry_max_attempts=retry_max_attempts,
+            send_failed_reports=send_failed_reports,
+            retry_notify_owners=retry_notify_owners,
+            retry_notify_recipients=retry_notify_recipients,
         )
     db.session.add(report_schedule)
     db.session.commit()
@@ -139,6 +149,11 @@ def create_report_notification(
     ccTarget: Optional[str] = None,  # noqa: N803
     bccTarget: Optional[str] = None,  # noqa: N803
     use_slack_v2: bool = False,
+    retry_on_failure: bool = False,
+    retry_max_attempts: int = 3,
+    send_failed_reports: bool = False,
+    retry_notify_owners: bool = True,
+    retry_notify_recipients: bool = False,
 ) -> ReportSchedule:
     if not editors:
         default_owner = (
@@ -188,6 +203,11 @@ def create_report_notification(
         report_format=report_format or ReportDataFormat.PNG,
         extra=extra,
         force_screenshot=force_screenshot,
+        retry_on_failure=retry_on_failure,
+        retry_max_attempts=retry_max_attempts,
+        send_failed_reports=send_failed_reports,
+        retry_notify_owners=retry_notify_owners,
+        retry_notify_recipients=retry_notify_recipients,
     )
     return report_schedule