fix(reports): give each Slack channel a retry budget

UPDATING.md

@@ -586,7 +586,7 @@ With the flag enabled: `DELETE /api/v1/chart/<id>` no longer hard-deletes the ch
 
 **Importer behavior:** importing a chart YAML whose UUID matches an existing **soft-deleted** chart is treated as an implicit restore-with-update — **and this happens even when `overwrite` is not set**. This is a deliberate asymmetry with active rows: an active chart imported without `overwrite=true` is returned unchanged, but a soft-deleted UUID match is restored *and* has the upload's contents applied regardless of the `overwrite` argument, on the reasoning that re-importing a deleted chart's exact UUID is an explicit request to bring it back. The restore preserves the original PK and all out-of-archive references (`dashboard_slices` junctions, `report.chart_id`, tag rows). The operation is permission-gated: non-editors get `ImportFailedError`, and callers without `can_write` get `ImportFailedError` instead of silently receiving the soft-deleted row.
 
-- [39914](https://github.com/apache/superset/pull/39914) `ALERT_REPORT_SLACK_V2` now defaults to `True` and the legacy Slack v1 integration (`Slack` recipient type, `files.upload` API) is deprecated for removal in the next major. Slack blocked new apps from `files.upload` in May 2024 and fully retired the method for all apps on November 12, 2025; because the v1 path sends files through `files.upload`, v1 file-bearing sends now fail at the API level — only text-only `chat_postMessage` still works via the legacy path. Grant your Slack bot the `channels:read` and `groups:read` scopes so existing `Slack` recipients can be auto-upgraded to `SlackV2` on next send. Operators who explicitly override the flag to `False`, or whose Slack bot is missing those scopes, will see deprecation warnings while text-only sends continue through the legacy path.
+- [39914](https://github.com/apache/superset/pull/39914) defaults `ALERT_REPORT_SLACK_V2` to `True` and deprecates legacy Slack v1; [42089](https://github.com/apache/superset/pull/42089) automatically upgrades resolvable v1 recipients, preserves text-only v1 delivery with execution warnings when migration cannot finish, and rejects retired v1 file uploads with actionable scope guidance. Grant bots `channels:read` and `groups:read` for public and private channel resolution. Slack delivery uses at-most-once terminal writes and an independent per-channel retry budget configured by `SLACK_SEND_RETRY_MAX_TIME`; see [Alerts and Reports](https://superset.apache.org/docs/configuration/alerts-reports#slack-delivery-timeouts-and-retries).
 
 ### Soft delete and restore for dashboards
 

docs/admin_docs/configuration/alerts-reports.mdx

@@ -96,6 +96,33 @@ SLACK_TEAM_ID = "T01234567"
 This defaults to `None` and only needs to be set when using an org-scoped token;
 it is accepted but ignored for standard workspace-level tokens.
 
+#### Slack delivery timeouts and retries
+
+Slack delivery uses a request timeout and an application retry budget:
+
+```python
+# Timeout for one Slack API request, in seconds
+SLACK_API_TIMEOUT = 30
+
+# Retry budget for each destination channel across its files and upload phases
+SLACK_SEND_RETRY_MAX_TIME = 150
+
+# Number of explicit HTTP 429 responses retried using Slack's Retry-After value
+SLACK_API_RATE_LIMIT_RETRY_COUNT = 2
+```
+
+Each channel receives an independent `SLACK_SEND_RETRY_MAX_TIME` budget, so a
+slow or unavailable channel does not reduce the time available to later
+channels. The effective per-channel budget is at least one second longer than
+`SLACK_API_TIMEOUT`. Increase it for large files or slow Slack workspaces.
+
+To avoid posting the same report twice, Superset does not replay terminal
+`chat.postMessage` or `files.completeUploadExternal` operations after ambiguous
+server or transport failures. Explicit Slack HTTP 429 responses remain
+retryable. These delivery settings and semantics apply to Slack v2 reports and
+legacy text-only Slack delivery, independently of the
+`ALERT_REPORT_SLACK_V2` feature flag.
+
 ### Webhook integration
 
 Superset can send alert and report notifications to any HTTP endpoint — useful for chat platforms, incident management tools, or custom automation.

superset/commands/report/execute.py

@@ -55,6 +55,7 @@ from superset.commands.report.exceptions import (
     ReportScheduleXlsxFailedError,
     ReportScheduleXlsxTimeout,
 )
+from superset.commands.report.slack_upgrade import SlackV1UpgradeCoordinator
 from superset.common.chart_data import ChartDataResultFormat, ChartDataResultType
 from superset.daos.report import (
     REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER,
@@ -79,18 +80,20 @@ from superset.reports.notifications.base import NotificationContent
 from superset.reports.notifications.exceptions import (
     NotificationError,
     NotificationParamException,
-    SlackV1NotificationError,
 )
+from superset.reports.notifications.slack import SlackNotification
 from superset.subjects.types import SubjectType
 from superset.tasks.utils import get_executor
 from superset.utils import json
-from superset.utils.core import HeaderDataType, override_user, recipients_string_to_list
+from superset.utils.core import HeaderDataType, override_user
 from superset.utils.csv import get_chart_csv_data, get_chart_dataframe
-from superset.utils.decorators import logs_context, transaction
+from superset.utils.decorators import (
+    logs_context,
+    transaction,
+)
 from superset.utils.file import sanitize_title
 from superset.utils.pdf import build_pdf_from_screenshots
 from superset.utils.screenshots import ChartScreenshot, DashboardScreenshot
-from superset.utils.slack import get_channels_with_search, SlackChannelTypes
 from superset.utils.urls import get_url_path
 
 if TYPE_CHECKING:
@@ -138,19 +141,29 @@ class BaseReportState:
         self._scheduled_dttm = scheduled_dttm
         self._start_dttm: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
         self._execution_id = execution_id
-        self._filter_warnings: list[str] = []
+        self._execution_warnings: list[str] = []
+        self._slack_v1_upgrade = SlackV1UpgradeCoordinator(
+            report_schedule,
+            execution_id,
+            self._execution_warnings,
+        )
 
     def update_report_schedule_and_log(
         self,
         state: ReportState,
         error_message: Optional[str] = None,
+        *,
+        include_execution_warnings: bool = True,
     ) -> None:
         """
         Update the report schedule state et al. and reflect the change in the execution
         log.
         """
         self.update_report_schedule(state)
-        self.create_log(error_message)
+        self.create_log(
+            error_message,
+            include_execution_warnings=include_execution_warnings,
+        )
 
     def update_report_schedule(self, state: ReportState) -> None:
         """
@@ -171,60 +184,15 @@ class BaseReportState:
         )
 
     def update_report_schedule_slack_v2(self) -> None:
-        """
-        Update the report schedule type and channels for all slack recipients to v2.
-        V2 uses ids instead of names for channels.
+        """Update every Slack v1 recipient atomically to Slack v2."""
+        self._slack_v1_upgrade.update_recipients()
 
-        Channel ids for every Slack recipient are resolved first and the
-        recipients are only mutated once all of them resolve. This keeps the
-        upgrade all-or-nothing: a single unresolvable channel can no longer
-        leave the schedule with some recipients already switched to v2 (and
-        persisted by a later error-log commit) while others are untouched.
-        """
-        resolved: list[tuple[ReportRecipients, str]] = []
-        try:
-            for recipient in self._report_schedule.recipients:
-                if recipient.type != ReportRecipientType.SLACK:
-                    continue
-                slack_recipients = json.loads(recipient.recipient_config_json)
-                # V1 method allowed to use leading `#` in the channel name
-                channel_names = (slack_recipients["target"] or "").replace("#", "")
-                # we need to ensure that existing reports can also fetch
-                # ids from private channels
-                channels = get_channels_with_search(
-                    search_string=channel_names,
-                    types=[
-                        SlackChannelTypes.PRIVATE,
-                        SlackChannelTypes.PUBLIC,
-                    ],
-                    exact_match=True,
-                )
-                channels_list = recipients_string_to_list(channel_names)
-                if len(channels_list) != len(channels):
-                    missing_channels = set(channels_list) - {
-                        channel["name"] for channel in channels
-                    }
-                    msg = (
-                        "Could not find the following channels: "
-                        f"{', '.join(missing_channels)}"
-                    )
-                    raise UpdateFailedError(msg)
-                channel_ids = ",".join(channel["id"] for channel in channels)
-                resolved.append((recipient, json.dumps({"target": channel_ids})))
-        except Exception as ex:
-            # No recipient has been mutated yet, so there is no partial upgrade
-            # to revert; surface the failure so the configuration can be fixed
-            # manually.
-            msg = f"Failed to update slack recipients to v2: {str(ex)}"
-            logger.exception(msg)
-            raise UpdateFailedError(msg) from ex
-
-        # Every Slack recipient resolved; apply the upgrade atomically.
-        for recipient, recipient_config_json in resolved:
-            recipient.type = ReportRecipientType.SLACKV2
-            recipient.recipient_config_json = recipient_config_json
-
-    def create_log(self, error_message: Optional[str] = None) -> None:
+    def create_log(
+        self,
+        error_message: Optional[str] = None,
+        *,
+        include_execution_warnings: bool = True,
+    ) -> None:
         """
         Creates a Report execution log, uses the current computed last_value for Alerts
 
@@ -241,6 +209,15 @@ class BaseReportState:
         """
         from sqlalchemy.orm.exc import StaleDataError
 
+        log_message: Optional[str]
+        if error_message == REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER:
+            log_message = error_message
+        else:
+            messages = [*self._execution_warnings] if include_execution_warnings else []
+            if error_message:
+                messages.append(error_message)
+            log_message = ";".join(messages) if messages else None
+
         try:
             # Reuse the in-flight WORKING trigger row for this execution, if any,
             # so a single execution surfaces as a single log entry.
@@ -267,7 +244,7 @@ class BaseReportState:
             log.value = self._report_schedule.last_value
             log.value_row_json = self._report_schedule.last_value_row_json
             log.state = self._report_schedule.last_state
-            log.error_message = error_message
+            log.error_message = log_message
             db.session.commit()  # pylint: disable=consider-using-transaction
         except StaleDataError as ex:
             # Report schedule was modified or deleted by another process
@@ -389,7 +366,7 @@ class BaseReportState:
                 self._report_schedule.get_native_filters_params()
             )
             if filter_warnings:
-                self._filter_warnings.extend(filter_warnings)
+                self._execution_warnings.extend(filter_warnings)
             if anchor := dashboard_state.get("anchor"):
                 try:
                     anchor_list = json.loads(anchor)
@@ -433,7 +410,7 @@ class BaseReportState:
             self._report_schedule.get_native_filters_params()
         )
         if filter_warnings:
-            self._filter_warnings.extend(filter_warnings)
+            self._execution_warnings.extend(filter_warnings)
         if native_filter_params and native_filter_params != "()":
             # Preserve any urlParams from extra.dashboard (e.g. standalone=true)
             # set via API even when ALERT_REPORT_TABS is off — same merge
@@ -1037,6 +1014,37 @@ class BaseReportState:
             header_data=header_data,
         )
 
+    def _send_notification(
+        self,
+        notification_content: NotificationContent,
+        recipient: ReportRecipients,
+    ) -> None:
+        """Send one notification, upgrading Slack v1 recipients when required."""
+        notification = create_notification(recipient, notification_content)
+        if app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"]:
+            logger.info(
+                "Would send notification for alert %s, to %s. "
+                "ALERT_REPORTS_NOTIFICATION_DRY_RUN is enabled, "
+                "set it to False to send notifications.",
+                self._report_schedule.name,
+                recipient.recipient_config_json,
+            )
+            return
+
+        if isinstance(notification, SlackNotification):
+            self._slack_v1_upgrade.send(
+                notification,
+                notification_content,
+                update_recipients=self.update_report_schedule_slack_v2,
+                create_upgraded_notification=lambda: create_notification(
+                    recipient,
+                    notification_content,
+                ),
+            )
+            return
+
+        notification.send()
+
     def _send(
         self,
         notification_content: NotificationContent,
@@ -1048,29 +1056,10 @@ class BaseReportState:
         :raises: CommandException
         """
         notification_errors: list[SupersetError] = []
+        self._slack_v1_upgrade.reset()
         for recipient in recipients:
-            notification = create_notification(recipient, notification_content)
             try:
-                try:
-                    if app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"]:
-                        logger.info(
-                            "Would send notification for alert %s, to %s. "
-                            "ALERT_REPORTS_NOTIFICATION_DRY_RUN is enabled, "
-                            "set it to False to send notifications.",
-                            self._report_schedule.name,
-                            recipient.recipient_config_json,
-                        )
-                    else:
-                        notification.send()
-                except SlackV1NotificationError as ex:
-                    # The slack notification should be sent with the v2 api
-                    logger.info(
-                        "Attempting to upgrade the report to Slackv2: %s", str(ex)
-                    )
-                    self.update_report_schedule_slack_v2()
-                    recipient.type = ReportRecipientType.SLACKV2
-                    notification = create_notification(recipient, notification_content)
-                    notification.send()
+                self._send_notification(notification_content, recipient)
             except (
                 UpdateFailedError,
                 NotificationParamException,
@@ -1212,13 +1201,7 @@ class ReportNotTriggeredErrorState(BaseReportState):
                     )
                     return
             self.send()
-            # Include filter warnings in the log if any were collected
-            warning_message = (
-                ";".join(self._filter_warnings) if self._filter_warnings else None
-            )
-            self.update_report_schedule_and_log(
-                ReportState.SUCCESS, error_message=warning_message
-            )
+            self.update_report_schedule_and_log(ReportState.SUCCESS, error_message=None)
         except (SupersetErrorsException, Exception) as first_ex:
             error_message = str(first_ex)
             if isinstance(first_ex, SupersetErrorsException):
@@ -1266,7 +1249,9 @@ class ReportNotTriggeredErrorState(BaseReportState):
                 finally:
                     try:
                         self.update_report_schedule_and_log(
-                            ReportState.ERROR, error_message=second_error_message
+                            ReportState.ERROR,
+                            error_message=second_error_message,
+                            include_execution_warnings=False,
                         )
                     except ReportScheduleUnexpectedError:
                         # Logging failed again, log it but don't let it hide first_ex
@@ -1399,13 +1384,7 @@ class ReportSuccessState(BaseReportState):
 
         try:
             self.send()
-            # Include filter warnings in the log if any were collected
-            warning_message = (
-                ";".join(self._filter_warnings) if self._filter_warnings else None
-            )
-            self.update_report_schedule_and_log(
-                ReportState.SUCCESS, error_message=warning_message
-            )
+            self.update_report_schedule_and_log(ReportState.SUCCESS, error_message=None)
         except Exception as ex:  # pylint: disable=broad-except
             try:
                 self.update_report_schedule_and_log(

superset/commands/report/slack_upgrade.py

@@ -0,0 +1,202 @@
+# 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 logging
+from collections.abc import Callable
+from uuid import UUID
+
+from flask import current_app as app
+
+from superset.commands.exceptions import UpdateFailedError
+from superset.reports.models import (
+    ReportRecipients,
+    ReportRecipientType,
+    ReportSchedule,
+)
+from superset.reports.notifications.base import BaseNotification, NotificationContent
+from superset.reports.notifications.exceptions import (
+    NotificationParamException,
+    SlackV1NotificationError,
+)
+from superset.reports.notifications.slack import (
+    SLACK_V1_FILE_UPLOAD_MESSAGE,
+    SlackNotification,
+)
+from superset.reports.notifications.slack_channel_resolver import (
+    resolve_slack_channel_ids,
+)
+from superset.utils import json
+from superset.utils.decorators import record_statsd_gauge_failure
+from superset.utils.slack import (
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
+    SlackChannelListingClientError,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class SlackV1UpgradeCoordinator:
+    """Coordinate one atomic Slack v1 upgrade and its per-recipient fallbacks."""
+
+    def __init__(
+        self,
+        report_schedule: ReportSchedule,
+        execution_id: UUID,
+        execution_warnings: list[str],
+    ) -> None:
+        self._report_schedule = report_schedule
+        self._execution_id = execution_id
+        self._execution_warnings = execution_warnings
+        self.reset()
+
+    def reset(self) -> None:
+        """Reset execution-scoped upgrade and fallback state."""
+        self._upgrade_error: NotificationParamException | UpdateFailedError | None = (
+            None
+        )
+        self._fallback_recorded = False
+
+    def update_recipients(self) -> None:
+        """Resolve and atomically convert every Slack v1 recipient to Slack v2."""
+        pending: list[tuple[ReportRecipients, list[str]]] = []
+        try:
+            for recipient in self._report_schedule.recipients:
+                if recipient.type != ReportRecipientType.SLACK:
+                    continue
+                try:
+                    slack_recipients = json.loads(recipient.recipient_config_json)
+                except (TypeError, ValueError) as ex:
+                    raise NotificationParamException(
+                        "Invalid Slack recipient configuration"
+                    ) from ex
+                target = (
+                    slack_recipients.get("target")
+                    if isinstance(slack_recipients, dict)
+                    else None
+                )
+                if not isinstance(target, str):
+                    raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+                channels = parse_slack_recipient_targets(target.replace("#", ""))
+                if not channels:
+                    raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+                pending.append((recipient, channels))
+
+            all_targets = list(
+                dict.fromkeys(
+                    channel for _, channels in pending for channel in channels
+                )
+            )
+            channel_ids = resolve_slack_channel_ids(all_targets) if all_targets else {}
+            resolved = [
+                (
+                    recipient,
+                    json.dumps(
+                        {
+                            "target": ",".join(
+                                channel_ids[channel] for channel in channels
+                            )
+                        }
+                    ),
+                )
+                for recipient, channels in pending
+            ]
+        except (NotificationParamException, SlackChannelListingClientError) as ex:
+            message = f"Failed to update slack recipients to v2: {ex}"
+            logger.warning(message)
+            raise NotificationParamException(message) from ex
+        except Exception as ex:
+            message = f"Failed to update slack recipients to v2: {ex}"
+            logger.exception(message)
+            raise UpdateFailedError(message) from ex
+
+        for recipient, recipient_config_json in resolved:
+            recipient.type = ReportRecipientType.SLACKV2
+            recipient.recipient_config_json = recipient_config_json
+
+    def send_fallback(
+        self,
+        notification: SlackNotification,
+        content: NotificationContent,
+        update_error: NotificationParamException | UpdateFailedError,
+    ) -> None:
+        """Deliver text through Slack v1 and record the first successful fallback."""
+        if content.has_attachments:
+            record_statsd_gauge_failure("reports.slack.send", update_error)
+            message = (
+                f"{SLACK_V1_FILE_UPLOAD_MESSAGE} "
+                f"Slack v2 upgrade failed: {update_error}"
+            )
+            if isinstance(update_error, UpdateFailedError):
+                raise UpdateFailedError(message) from update_error
+            raise NotificationParamException(message) from update_error
+
+        notification.send_legacy_text()
+        if self._fallback_recorded:
+            return
+
+        self._execution_warnings.append(
+            "Slack v2 upgrade unavailable; delivered the text-only report "
+            f"through deprecated Slack v1: {update_error}"
+        )
+        app.config["STATS_LOGGER"].incr("reports.slack.v1_fallback")
+        if isinstance(update_error, UpdateFailedError):
+            app.config["STATS_LOGGER"].incr("reports.slack.v1_fallback.system_error")
+            logger.error(
+                "Slack v2 upgrade failed with a system error; delivered the "
+                "text-only report through Slack v1 for this execution: %s",
+                update_error,
+                extra={
+                    "execution_id": self._execution_id,
+                    "report_schedule_id": self._report_schedule.id,
+                },
+            )
+        else:
+            logger.warning(
+                "Slack v2 upgrade unavailable; delivered the text-only report "
+                "through Slack v1 for this execution: %s",
+                update_error,
+            )
+        self._fallback_recorded = True
+
+    def send(
+        self,
+        notification: SlackNotification,
+        content: NotificationContent,
+        *,
+        update_recipients: Callable[[], None],
+        create_upgraded_notification: Callable[[], BaseNotification],
+    ) -> None:
+        """Send one Slack v1 recipient, upgrading or falling back when required."""
+        if self._upgrade_error is not None:
+            self.send_fallback(notification, content, self._upgrade_error)
+            return
+
+        try:
+            notification.send()
+        except SlackV1NotificationError as ex:
+            logger.info("Attempting to upgrade the report to Slackv2: %s", ex)
+            try:
+                update_recipients()
+            except (
+                NotificationParamException,
+                UpdateFailedError,
+            ) as update_error:
+                self._upgrade_error = update_error
+                self.send_fallback(notification, content, update_error)
+            else:
+                create_upgraded_notification().send()

superset/config.py

@@ -2439,6 +2439,12 @@ SLACK_API_RATE_LIMIT_RETRY_COUNT = 2
 # patching code, consistent with the SMTP/CSV/screenshot timeouts.
 SLACK_API_TIMEOUT = 30
 
+# Application retry budget (in seconds) for each Slack channel across all files
+# and upload phases. Increase this for slow channels or large-file reports. Each
+# channel receives an independent budget, and the effective value is always at
+# least one second longer than SLACK_API_TIMEOUT.
+SLACK_SEND_RETRY_MAX_TIME = 150
+
 # The webdriver to use for generating reports when using Selenium (not Playwright).
 # This setting is ignored when PLAYWRIGHT_REPORTS_AND_THUMBNAILS is enabled, as
 # Playwright always uses Chromium regardless of this value.

superset/reports/notifications/base.py

@@ -36,6 +36,11 @@ class NotificationContent:
     url: Optional[str] = None  # url to chart/dashboard for this screenshot
     embedded_data: Optional[pd.DataFrame] = None
 
+    @property
+    def has_attachments(self) -> bool:
+        """Return whether the notification contains any file attachment."""
+        return bool(self.csv or self.xlsx or self.pdf or self.screenshots)
+
 
 class BaseNotification:  # pylint: disable=too-few-public-methods
     """

superset/reports/notifications/exceptions.py

@@ -24,6 +24,10 @@ class NotificationError(SupersetException):
     """
 
 
+class NotificationTransientError(NotificationError):
+    """Temporary third-party delivery failure that should be retried later."""
+
+
 class SlackV1NotificationError(SupersetException):
     """
     Report should not be run with the slack v1 api

superset/reports/notifications/slack.py

@@ -15,18 +15,13 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from collections.abc import Sequence
-from io import IOBase
-from typing import Union
 
-import backoff
 from flask import g
+from slack_sdk import WebClient
 from slack_sdk.errors import (
     BotUserAccessError,
-    SlackApiError,
     SlackClientConfigurationError,
     SlackClientError,
-    SlackClientNotConnectedError,
     SlackObjectFormationError,
     SlackRequestError,
     SlackTokenRotationError,
@@ -42,23 +37,35 @@ from superset.reports.notifications.exceptions import (
     SlackV1NotificationError,
 )
 from superset.reports.notifications.slack_mixin import SlackMixin
+from superset.reports.notifications.slack_transport import (
+    call_slack_api_with_timeout,
+    send_to_slack_channels,
+)
 from superset.utils import json
-from superset.utils.core import recipients_string_to_list
 from superset.utils.decorators import statsd_gauge
 from superset.utils.slack import (
     get_slack_client,
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
     should_use_v2_api,
 )
 
 logger = logging.getLogger(__name__)
 
+SLACK_V1_FILE_UPLOAD_MESSAGE = (
+    "Slack v1 file uploads are no longer supported because Slack retired "
+    "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+    "both the `channels:read` and `groups:read` scopes so the recipient can "
+    "be upgraded to Slack v2."
+)
+
 
 # Deprecated: Slack v1 will be removed in the next major release. The Slack
 # `files.upload` endpoint was retired in 2025, so file-bearing sends already
-# fail at the API level; only text-only `chat_postMessage` sends still work
-# here. When the Slack bot has the `channels:read` and `groups:read` scopes,
-# existing v1 recipients are auto-upgraded to SlackV2 on first send via
-# `update_report_schedule_slack_v2`.
+# fail before attempting the retired v1 upload; only text-only
+# `chat_postMessage` sends still work here. When the Slack bot has the
+# `channels:read` and `groups:read` scopes, existing v1 recipients are
+# auto-upgraded to SlackV2 on their first eligible send.
 class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
     """
     Sends a slack notification for a report recipient
@@ -66,58 +73,49 @@ class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-fe
 
     type = ReportRecipientType.SLACK
 
-    def _get_channel(self) -> str:
+    def _get_channels(self) -> list[str]:
         """
-        Get the recipient's channel(s).
-        Note Slack SDK uses "channel" to refer to one or more
-        channels. Multiple channels are demarcated by a comma.
-        :returns: The comma separated list of channel(s)
+        Get the recipient's normalized channel list.
+
+        :returns: channel names with duplicates removed in configured order
         """
-        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        try:
+            recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        except (KeyError, TypeError, ValueError) as ex:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE) from ex
 
-        return ",".join(recipients_string_to_list(recipient_str))
+        if not isinstance(recipient_str, str):
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
 
-    def _get_inline_files(
-        self,
-    ) -> tuple[Union[str, None], Sequence[Union[str, IOBase, bytes]]]:
-        if self._content.csv:
-            return ("csv", [self._content.csv])
-        if self._content.xlsx:
-            return ("xlsx", [self._content.xlsx])
-        if self._content.screenshots:
-            return ("png", self._content.screenshots)
-        if self._content.pdf:
-            return ("pdf", [self._content.pdf])
-        return (None, [])
+        return parse_slack_recipient_targets(recipient_str)
+
+    @staticmethod
+    def _send_text(client: WebClient, channels: list[str], body: str) -> None:
+        """Send a text notification once to each configured channel."""
+        if not channels:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+        send_to_slack_channels(
+            channels,
+            lambda target, retry_deadline: call_slack_api_with_timeout(
+                client,
+                client.chat_postMessage,
+                retry_deadline=retry_deadline,
+                retry_transient_errors=False,
+                channel=target,
+                text=body,
+            ),
+        )
+
+    def _send_legacy_text(self) -> None:
+        if self._content.has_attachments:
+            raise NotificationParamException(SLACK_V1_FILE_UPLOAD_MESSAGE)
 
-    @backoff.on_exception(backoff.expo, SlackApiError, factor=10, base=2, max_tries=5)
-    @statsd_gauge("reports.slack.send")
-    def send(self) -> None:
-        file_type, files = self._get_inline_files()
-        title = self._content.name
         body = self._get_body(content=self._content)
         global_logs_context = getattr(g, "logs_context", {}) or {}
-
-        # see if the v2 api will work
-        if should_use_v2_api():
-            # if we can fetch channels, then raise an error and use the v2 api
-            raise SlackV1NotificationError
-
         try:
-            client = get_slack_client()
-            channel = self._get_channel()
-            # files_upload returns SlackResponse as we run it in sync mode.
-            if files:
-                for file in files:
-                    client.files_upload(
-                        channels=channel,
-                        file=file,
-                        initial_comment=body,
-                        title=title,
-                        filetype=file_type,
-                    )
-            else:
-                client.chat_postMessage(channel=channel, text=body)
+            client = get_slack_client(for_delivery=True)
+            channels = self._get_channels()
+            self._send_text(client, channels, body)
             logger.info(
                 "Report sent to slack",
                 extra={
@@ -134,9 +132,21 @@ class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-fe
             raise NotificationMalformedException(str(ex)) from ex
         except SlackTokenRotationError as ex:
             raise NotificationAuthorizationException(str(ex)) from ex
-        except (SlackClientNotConnectedError, SlackApiError) as ex:
-            raise NotificationUnprocessableException(str(ex)) from ex
         except SlackClientError as ex:
-            # this is the base class for all slack client errors
-            # keep it last so that it doesn't interfere with @backoff
+            # SlackClientError is the base class; keep it last so subclasses
+            # retain their more specific notification classification.
             raise NotificationUnprocessableException(str(ex)) from ex
+
+    @statsd_gauge("reports.slack.send")
+    def send_legacy_text(self) -> None:
+        """Send through Slack v1 without repeating the v2 availability probe."""
+        self._send_legacy_text()
+
+    @statsd_gauge(
+        "reports.slack.send",
+        ignored_exceptions=(SlackV1NotificationError,),
+    )
+    def send(self) -> None:
+        if should_use_v2_api(raise_on_error=self._content.has_attachments):
+            raise SlackV1NotificationError
+        self._send_legacy_text()

superset/reports/notifications/slack_channel_resolver.py

@@ -0,0 +1,104 @@
+# 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.
+
+from superset.reports.notifications.exceptions import NotificationParamException
+from superset.utils.slack import (
+    get_channels_with_search,
+    refresh_cached_slack_channels_with_search,
+    SlackChannel,
+    SlackChannelTypes,
+)
+
+
+def _get_slack_channels(
+    search_string: str,
+) -> tuple[list[SlackChannel], bool]:
+    """Fetch matching Slack channels and same-read cache provenance."""
+    channels, used_cache = get_channels_with_search(
+        search_string=search_string,
+        types=[
+            SlackChannelTypes.PRIVATE,
+            SlackChannelTypes.PUBLIC,
+        ],
+        exact_match=True,
+        force=False,
+        return_cache_status=True,
+    )
+    return channels, used_cache
+
+
+def _match_slack_channel(
+    target: str,
+    channels: list[SlackChannel],
+) -> SlackChannel | None:
+    """Resolve one target with deterministic exact-ID, name, then folded-ID order."""
+    match_groups = (
+        [channel for channel in channels if channel["id"] == target],
+        [
+            channel
+            for channel in channels
+            if channel["name"].casefold() == target.casefold()
+        ],
+        [
+            channel
+            for channel in channels
+            if channel["id"].casefold() == target.casefold()
+        ],
+    )
+    for matches in match_groups:
+        if len(matches) > 1:
+            raise NotificationParamException(
+                f"Slack channel target is ambiguous: {target}"
+            )
+        if matches:
+            return matches[0]
+    return None
+
+
+def _match_slack_channels(
+    targets: list[str],
+    channels: list[SlackChannel],
+) -> tuple[dict[str, SlackChannel], list[str]]:
+    resolved: dict[str, SlackChannel] = {}
+    missing: list[str] = []
+    for target in targets:
+        if channel := _match_slack_channel(target, channels):
+            resolved[target] = channel
+        else:
+            missing.append(target)
+    return resolved, missing
+
+
+def resolve_slack_channel_ids(
+    targets: list[str],
+) -> dict[str, str]:
+    """Resolve Slack names or IDs, refreshing only a stale cached listing."""
+    search_string = ",".join(targets)
+    channels, used_cached_channels = _get_slack_channels(search_string)
+    channels_by_target, missing_channels = _match_slack_channels(targets, channels)
+    if missing_channels and used_cached_channels:
+        channels = refresh_cached_slack_channels_with_search(
+            search_string=search_string,
+            types=[SlackChannelTypes.PRIVATE, SlackChannelTypes.PUBLIC],
+            exact_match=True,
+        )
+        channels_by_target, missing_channels = _match_slack_channels(targets, channels)
+    if missing_channels:
+        raise NotificationParamException(
+            f"Could not find the following channels: {', '.join(missing_channels)}"
+        )
+    return {target: channel["id"] for target, channel in channels_by_target.items()}

superset/reports/notifications/slack_transport.py

@@ -0,0 +1,269 @@
+# 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 math
+import time
+from collections.abc import Callable
+from functools import partial
+from typing import TypeVar
+from urllib.error import HTTPError
+
+import backoff
+from flask import current_app as app
+from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError, SlackRequestError
+
+from superset.reports.notifications.exceptions import (
+    NotificationTransientError,
+    NotificationUnprocessableException,
+)
+from superset.utils.slack import (
+    get_slack_api_error_code,
+    get_slack_api_status_code,
+    is_retryable_slack_transport_error,
+    is_transient_slack_api_error,
+    is_transient_slack_transport_error,
+    SLACK_TRANSIENT_TRANSPORT_ERRORS,
+)
+
+SLACK_API_TIMEOUT_MARGIN = 1
+
+_SlackApiResult = TypeVar("_SlackApiResult")
+
+_SLACK_RETRY_DEADLINE_MESSAGE = (
+    "Slack send retry deadline exceeded; increase SLACK_SEND_RETRY_MAX_TIME "
+    "for slow-channel or large-file reports"
+)
+
+
+class SlackRetryDeadlineError(Exception):
+    """A Slack operation was skipped because the shared send budget expired."""
+
+    def __init__(self) -> None:
+        super().__init__(_SLACK_RETRY_DEADLINE_MESSAGE)
+
+
+class SlackChannelResponseError(SlackRequestError):
+    """Slack returned malformed channel-specific data before a terminal write."""
+
+
+_SLACK_RETRY_ERRORS = (SlackApiError, *SLACK_TRANSIENT_TRANSPORT_ERRORS)
+_SLACK_CHANNEL_FAILURES = (
+    *_SLACK_RETRY_ERRORS,
+    SlackChannelResponseError,
+    SlackRetryDeadlineError,
+)
+
+
+def _get_slack_send_retry_max_time() -> float:
+    """Return the effective per-channel send budget in seconds."""
+    configured_budget = float(app.config["SLACK_SEND_RETRY_MAX_TIME"])
+    request_timeout = float(app.config.get("SLACK_API_TIMEOUT", 30))
+    return max(
+        configured_budget,
+        request_timeout + SLACK_API_TIMEOUT_MARGIN,
+    )
+
+
+def _give_up_slack_api_retry(
+    ex: Exception,
+    *,
+    retry_transient_errors: bool = True,
+    retry_transport_errors: bool = False,
+) -> bool:
+    """Return whether application backoff should stop retrying a Slack call."""
+    if isinstance(ex, HTTPError) and ex.code == 429:
+        return True
+    if not isinstance(ex, SlackApiError):
+        if not retry_transient_errors:
+            return True
+        return not (
+            is_transient_slack_transport_error(ex)
+            if retry_transport_errors
+            else is_retryable_slack_transport_error(ex)
+        )
+
+    status_code = get_slack_api_status_code(ex)
+    # call_slack_api handles HTTP 429 within the shared monotonic deadline.
+    # Retrying an exhausted 429 through the outer backoff would multiply the
+    # operator-configured rate-limit budget.
+    error_code = get_slack_api_error_code(ex)
+    if status_code == 429:
+        return True
+    if not retry_transient_errors:
+        return True
+    if is_transient_slack_api_error(ex, error_code):
+        return False
+    if status_code is not None and 400 <= status_code < 500:
+        return True
+    return bool(error_code)
+
+
+def _get_slack_retry_after(ex: SlackApiError | HTTPError) -> float | None:
+    response = getattr(ex, "response", ex)
+    headers = getattr(response, "headers", None)
+    if headers is None:
+        return None
+    for name in headers.keys():
+        if name.lower() != "retry-after":
+            continue
+        value = headers.get(name)
+        raw_value = value[0] if isinstance(value, list) else value
+        try:
+            retry_after = float(raw_value)
+            return max(retry_after, 0.0) if math.isfinite(retry_after) else None
+        except (TypeError, ValueError):
+            return None
+    return None
+
+
+def _get_slack_rate_limit_status(ex: SlackApiError | HTTPError) -> int | None:
+    if isinstance(ex, HTTPError):
+        return ex.code
+    return get_slack_api_status_code(ex)
+
+
+def call_slack_api(
+    method: Callable[..., _SlackApiResult],
+    *,
+    retry_deadline: float | None = None,
+    retry_transient_errors: bool = True,
+    retry_transport_errors: bool = False,
+    retry_rate_limits: bool = True,
+    **kwargs: object,
+) -> _SlackApiResult:
+    """Call Slack with bounded retries, optionally sharing an outer deadline."""
+    if retry_deadline is None:
+        retry_deadline = time.monotonic() + _get_slack_send_retry_max_time()
+    max_time = retry_deadline - time.monotonic()
+    if max_time <= 0:
+        raise SlackRetryDeadlineError
+
+    max_rate_limit_retries = (
+        max(
+            int(app.config.get("SLACK_API_RATE_LIMIT_RETRY_COUNT", 2)),
+            0,
+        )
+        if retry_rate_limits
+        else 0
+    )
+    rate_limit_retries = 0
+
+    @backoff.on_exception(
+        backoff.expo,
+        _SLACK_RETRY_ERRORS,
+        factor=10,
+        base=2,
+        max_tries=5,
+        max_time=max_time,
+        giveup=partial(
+            _give_up_slack_api_retry,
+            retry_transient_errors=retry_transient_errors,
+            retry_transport_errors=retry_transport_errors,
+        ),
+    )
+    def call() -> _SlackApiResult:
+        nonlocal rate_limit_retries
+        while True:
+            if time.monotonic() >= retry_deadline:
+                raise SlackRetryDeadlineError
+            try:
+                return method(**kwargs)
+            except (SlackApiError, HTTPError) as ex:
+                if (
+                    _get_slack_rate_limit_status(ex) != 429
+                    or rate_limit_retries >= max_rate_limit_retries
+                ):
+                    raise
+                retry_after = _get_slack_retry_after(ex)
+                if retry_after is None:
+                    raise
+                remaining = retry_deadline - time.monotonic()
+                if retry_after >= remaining:
+                    raise SlackRetryDeadlineError from ex
+                time.sleep(retry_after)
+                rate_limit_retries += 1
+
+    return call()
+
+
+def call_slack_api_with_timeout(
+    client: WebClient,
+    method: Callable[..., _SlackApiResult],
+    *,
+    retry_deadline: float,
+    retry_transient_errors: bool = True,
+    retry_transport_errors: bool = False,
+    retry_rate_limits: bool = True,
+    **kwargs: object,
+) -> _SlackApiResult:
+    """Call Slack with the SDK request timeout capped by the shared budget."""
+    original_timeout = client.timeout
+
+    def call() -> _SlackApiResult:
+        remaining = retry_deadline - time.monotonic()
+        if remaining <= 0:
+            raise SlackRetryDeadlineError
+        client.timeout = min(float(original_timeout), remaining)
+        try:
+            return method(**kwargs)
+        finally:
+            client.timeout = original_timeout
+
+    return call_slack_api(
+        call,
+        retry_deadline=retry_deadline,
+        retry_transient_errors=retry_transient_errors,
+        retry_transport_errors=retry_transport_errors,
+        retry_rate_limits=retry_rate_limits,
+    )
+
+
+def send_to_slack_channels(
+    channels: list[str],
+    send_to_channel: Callable[[str, float], None],
+) -> None:
+    """Send to each channel with an independent application retry deadline."""
+    retry_max_time = _get_slack_send_retry_max_time()
+    failures: list[tuple[str, Exception]] = []
+    for channel in channels:
+        channel_deadline = time.monotonic() + retry_max_time
+        try:
+            send_to_channel(channel, channel_deadline)
+        except _SLACK_CHANNEL_FAILURES as ex:
+            failures.append((channel, ex))
+
+    if not failures:
+        return
+
+    details = "; ".join(f"{channel}: {error}" for channel, error in failures)
+    message = f"Slack delivery failed for the following channels: {details}"
+    if any(
+        isinstance(error, SlackRetryDeadlineError)
+        or isinstance(error, SlackChannelResponseError)
+        or is_transient_slack_transport_error(error)
+        or (
+            isinstance(error, SlackApiError)
+            and is_transient_slack_api_error(
+                error,
+                get_slack_api_error_code(error),
+            )
+        )
+        for _, error in failures
+    ):
+        raise NotificationTransientError(message) from failures[0][1]
+    raise NotificationUnprocessableException(message) from failures[0][1]

superset/reports/notifications/slackv2.py

@@ -15,18 +15,17 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from collections.abc import Callable, Sequence
-from io import IOBase
-from typing import List, Union
+import time
+from email.message import Message
+from typing import List
+from urllib.error import HTTPError
 
-import backoff
 from flask import g
+from slack_sdk import WebClient
 from slack_sdk.errors import (
     BotUserAccessError,
-    SlackApiError,
     SlackClientConfigurationError,
     SlackClientError,
-    SlackClientNotConnectedError,
     SlackObjectFormationError,
     SlackRequestError,
     SlackTokenRotationError,
@@ -41,61 +40,92 @@ from superset.reports.notifications.exceptions import (
     NotificationUnprocessableException,
 )
 from superset.reports.notifications.slack_mixin import SlackMixin
+from superset.reports.notifications.slack_transport import (
+    call_slack_api,
+    call_slack_api_with_timeout,
+    send_to_slack_channels,
+    SlackChannelResponseError,
+    SlackRetryDeadlineError,
+)
 from superset.utils import json
-from superset.utils.core import recipients_string_to_list
 from superset.utils.decorators import statsd_gauge
-from superset.utils.slack import get_slack_client
+from superset.utils.slack import (
+    get_slack_client,
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
+)
 
 logger = logging.getLogger(__name__)
 
-_TRANSIENT_SLACK_API_ERROR_CODES = frozenset(
-    {
-        "fatal_error",
-        "internal_error",
-        "ratelimited",
-        "request_timeout",
-        "rollup_error",
-        "service_unavailable",
-        "timeout",
-    }
-)
 
+def _upload_file_to_slack(
+    client: WebClient,
+    *,
+    channel: str,
+    file: bytes,
+    initial_comment: str,
+    title: str,
+    filename: str,
+    retry_deadline: float,
+) -> None:
+    """Upload one file without replaying completed phases during retries."""
+    data = file
+    upload_url_response = call_slack_api_with_timeout(
+        client,
+        client.files_getUploadURLExternal,
+        retry_deadline=retry_deadline,
+        retry_transport_errors=True,
+        filename=filename,
+        length=len(data),
+    )
+    try:
+        file_id = upload_url_response.get("file_id")
+        upload_url = upload_url_response.get("upload_url")
+    except (AttributeError, TypeError) as ex:
+        raise SlackChannelResponseError(
+            "Slack did not return valid upload metadata"
+        ) from ex
+    if (
+        not isinstance(file_id, str)
+        or not file_id
+        or not isinstance(upload_url, str)
+        or not upload_url
+    ):
+        raise SlackChannelResponseError("Slack did not return a file ID and upload URL")
 
-def _get_slack_api_error_code(ex: SlackApiError) -> str:
-    response = getattr(ex, "response", None)
-    data = getattr(response, "data", None)
-    if not isinstance(data, dict):
-        data = response if isinstance(response, dict) else {}
-    return str(data.get("error") or "")
+    def upload_file() -> None:
+        remaining = retry_deadline - time.monotonic()
+        if remaining <= 0:
+            raise SlackRetryDeadlineError
+        # files_upload_v2 uses this SDK helper internally. Calling it directly
+        # keeps retries scoped to the raw upload rather than replaying URL creation.
+        result = client._upload_file(  # pylint: disable=protected-access
+            url=upload_url,
+            data=data,
+            logger=logger,
+            timeout=min(float(client.timeout), remaining),
+            proxy=client.proxy,
+            ssl=client.ssl,
+        )
+        if result.status != 200:
+            raise HTTPError(
+                upload_url,
+                result.status,
+                f"Slack external upload failed: {result.body}",
+                Message(),
+                None,
+            )
 
-
-def _get_slack_api_status_code(ex: SlackApiError) -> int | None:
-    response = getattr(ex, "response", None)
-    return getattr(response, "status_code", None)
-
-
-def _give_up_slack_api_retry(ex: Exception) -> bool:
-    if not isinstance(ex, SlackApiError):
-        return False
-
-    status_code = _get_slack_api_status_code(ex)
-    if status_code == 429 or (status_code is not None and 500 <= status_code < 600):
-        return False
-
-    error_code = _get_slack_api_error_code(ex)
-    return bool(error_code and error_code not in _TRANSIENT_SLACK_API_ERROR_CODES)
-
-
-@backoff.on_exception(
-    backoff.expo,
-    (SlackApiError, SlackClientNotConnectedError),
-    factor=10,
-    base=2,
-    max_tries=5,
-    giveup=_give_up_slack_api_retry,
-)
-def _call_slack_api(method: Callable[..., object], **kwargs: object) -> None:
-    method(**kwargs)
+    call_slack_api(upload_file, retry_deadline=retry_deadline)
+    call_slack_api_with_timeout(
+        client,
+        client.files_completeUploadExternal,
+        retry_deadline=retry_deadline,
+        retry_transient_errors=False,
+        files=[{"id": file_id, "title": title}],
+        channel_id=channel,
+        initial_comment=initial_comment,
+    )
 
 
 class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
@@ -111,13 +141,19 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
         :returns: A list of channel ids: "EID676L"
         :raises NotificationParamException or SlackApiError: If the recipient is not found
         """  # noqa: E501
-        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        try:
+            recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        except (KeyError, TypeError, ValueError) as ex:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE) from ex
 
-        return recipients_string_to_list(recipient_str)
+        if not isinstance(recipient_str, str):
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+
+        return parse_slack_recipient_targets(recipient_str)
 
     def _get_inline_files(
         self,
-    ) -> tuple[Union[str, None], Sequence[Union[str, IOBase, bytes]]]:
+    ) -> tuple[str | None, list[bytes]]:
         if self._content.csv:
             return ("csv", [self._content.csv])
         if self._content.xlsx:
@@ -132,24 +168,28 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
     def send(self) -> None:
         global_logs_context = getattr(g, "logs_context", {}) or {}
         try:
-            client = get_slack_client()
+            client = get_slack_client(for_delivery=True)
             title = self._content.name
             body = self._get_body(content=self._content)
 
             channels = self._get_channels()
 
             if not channels:
-                raise NotificationParamException("No recipients saved in the report")
+                raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
 
             file_type, files = self._get_inline_files()
-            file_name = f"{title}.{file_type}"
 
-            # files_upload returns SlackResponse as we run it in sync mode.
-            for channel in channels:
+            def send_to_channel(channel: str, retry_deadline: float) -> None:
                 if len(files) > 0:
+                    if file_type is None:
+                        raise SlackChannelResponseError(
+                            "Slack upload file type was not provided"
+                        )
+                    file_name = f"{title}.{file_type}"
                     for file in files:
-                        _call_slack_api(
-                            client.files_upload_v2,
+                        _upload_file_to_slack(
+                            client,
+                            retry_deadline=retry_deadline,
                             channel=channel,
                             file=file,
                             initial_comment=body,
@@ -157,7 +197,16 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
                             filename=file_name,
                         )
                 else:
-                    _call_slack_api(client.chat_postMessage, channel=channel, text=body)
+                    call_slack_api_with_timeout(
+                        client,
+                        client.chat_postMessage,
+                        retry_deadline=retry_deadline,
+                        retry_transient_errors=False,
+                        channel=channel,
+                        text=body,
+                    )
+
+            send_to_slack_channels(channels, send_to_channel)
 
             logger.info(
                 "Report sent to slack",
@@ -175,9 +224,7 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
             raise NotificationMalformedException(str(ex)) from ex
         except SlackTokenRotationError as ex:
             raise NotificationAuthorizationException(str(ex)) from ex
-        except (SlackClientNotConnectedError, SlackApiError) as ex:
-            raise NotificationUnprocessableException(str(ex)) from ex
         except SlackClientError as ex:
-            # this is the base class for all slack client errors
-            # keep it last so that it doesn't interfere with @backoff
+            # SlackClientError is the base class; keep it last so subclasses
+            # retain their more specific notification classification.
             raise NotificationUnprocessableException(str(ex)) from ex

superset/utils/decorators.py

@@ -36,7 +36,21 @@ if TYPE_CHECKING:
     from superset.stats_logger import BaseStatsLogger
 
 
-def statsd_gauge(metric_prefix: str | None = None) -> Callable[..., Any]:
+def record_statsd_gauge_failure(metric_prefix: str, ex: Exception) -> None:
+    """Record a warning or error gauge using the shared exception contract."""
+    suffix = (
+        "warning"
+        if hasattr(ex, "status") and ex.status < 500  # pylint: disable=no-member
+        else "error"
+    )
+    app.config["STATS_LOGGER"].gauge(f"{metric_prefix}.{suffix}", 1)
+
+
+def statsd_gauge(
+    metric_prefix: str | None = None,
+    *,
+    ignored_exceptions: tuple[type[Exception], ...] = (),
+) -> Callable[..., Any]:
     def decorate(f: Callable[..., Any]) -> Callable[..., Any]:
         """
         Handle sending statsd gauge metric from any method or function
@@ -48,13 +62,10 @@ def statsd_gauge(metric_prefix: str | None = None) -> Callable[..., Any]:
                 result = f(*args, **kwargs)
                 app.config["STATS_LOGGER"].gauge(f"{metric_prefix_}.ok", 1)
                 return result
+            except ignored_exceptions:
+                raise
             except Exception as ex:
-                if (
-                    hasattr(ex, "status") and ex.status < 500  # pylint: disable=no-member
-                ):
-                    app.config["STATS_LOGGER"].gauge(f"{metric_prefix_}.warning", 1)
-                else:
-                    app.config["STATS_LOGGER"].gauge(f"{metric_prefix_}.error", 1)
+                record_statsd_gauge_failure(metric_prefix_, ex)
                 raise
 
         return wrapped

superset/utils/slack.py

@@ -19,16 +19,32 @@
 import functools
 import logging
 import warnings
-from typing import Any, Callable, Optional
+from http.client import RemoteDisconnected
+from typing import (
+    Any,
+    Callable,
+    Literal,
+    NotRequired,
+    Optional,
+    overload,
+    TypedDict,
+)
+from urllib.error import HTTPError, URLError
 
 from flask import current_app as app
 from slack_sdk import WebClient
-from slack_sdk.errors import SlackApiError, SlackClientError as SlackSDKClientError
+from slack_sdk.errors import (
+    SlackApiError,
+    SlackClientError as SlackSDKClientError,
+    SlackClientNotConnectedError,
+)
 from slack_sdk.http_retry.builtin_handlers import RateLimitErrorRetryHandler
 
 from superset import feature_flag_manager
+from superset.constants import CACHE_DISABLED_TIMEOUT
 from superset.exceptions import SupersetException
 from superset.extensions import cache_manager
+from superset.extensions.metastore_cache import SupersetMetastoreCache
 from superset.reports.schemas import SlackChannelSchema
 from superset.utils import cache as cache_util
 from superset.utils.backports import StrEnum
@@ -47,10 +63,9 @@ _SLACK_V1_DEPRECATION_MESSAGE = (
 )
 
 
-# functools.cache gives us a process-lifetime, thread-safe one-shot guard
-# without the read-then-write race that bare module globals would have under
-# multi-threaded WSGI workers. The cached return value (None) is irrelevant —
-# we only care that the body executes at most once per process.
+# functools.cache suppresses repeated calls after the first one completes.
+# Concurrent first calls may both emit, which is acceptable for a deprecation
+# warning and avoids managing additional process-local synchronization.
 @functools.cache
 def _emit_v1_flag_off_deprecation() -> None:
     warnings.warn(_SLACK_V1_DEPRECATION_MESSAGE, DeprecationWarning, stacklevel=3)
@@ -72,11 +87,123 @@ class SlackChannelTypes(StrEnum):
 _SLACK_CONVERSATION_TYPES = ",".join(SlackChannelTypes)
 
 
+class SlackChannel(TypedDict):
+    """Normalized Slack channel fields used for report-recipient resolution."""
+
+    id: str
+    name: str
+    is_private: bool
+    is_member: NotRequired[bool]
+
+
 class SlackClientError(Exception):
     pass
 
 
-def get_slack_client() -> WebClient:
+class SlackV2ProbeError(SupersetException):
+    """Transient probe failure classified by ``_send`` as a system error.
+
+    ``BaseReportState._send`` catches this through ``SupersetException`` and
+    uses the inherited status code to select ERROR severity.
+    """
+
+
+class SlackV2ProbeClientError(SlackV2ProbeError):
+    """Permanent probe failure classified by ``_send`` as a client warning.
+
+    ``BaseReportState._send`` catches this through ``SupersetException`` and
+    uses this status code to select WARNING severity.
+    """
+
+    status = 422
+
+
+class SlackChannelListingError(SupersetException):
+    """Slack channel listing failed due to a transient service condition."""
+
+
+class SlackChannelListingClientError(SlackChannelListingError):
+    """Slack channel listing failed due to permanent token or client setup."""
+
+    status = 422
+
+
+_TRANSIENT_SLACK_API_ERROR_CODES = frozenset(
+    {
+        "fatal_error",
+        "internal_error",
+        "ratelimited",
+        "request_timeout",
+        "rollup_error",
+        "service_unavailable",
+        "timeout",
+    }
+)
+
+SLACK_TRANSIENT_TRANSPORT_ERRORS: tuple[type[Exception], ...] = (
+    SlackClientNotConnectedError,
+    URLError,
+    ConnectionResetError,
+    RemoteDisconnected,
+    TimeoutError,
+)
+
+_SLACK_CHANNEL_REFRESH_COOLDOWN_SECONDS = 300
+
+
+NO_SLACK_RECIPIENTS_MESSAGE = "No recipients saved in the report"
+
+
+def parse_slack_recipient_targets(target: str) -> list[str]:
+    """Parse Slack targets, removing duplicates while preserving their order."""
+    return list(dict.fromkeys(recipients_string_to_list(target)))
+
+
+def get_slack_api_error_data(ex: SlackApiError) -> dict[str, Any]:
+    """Return a Slack API error payload across SDK response representations."""
+    response = getattr(ex, "response", None)
+    data = getattr(response, "data", None)
+    if not isinstance(data, dict):
+        data = response if isinstance(response, dict) else {}
+    return data
+
+
+def get_slack_api_error_code(ex: SlackApiError) -> str:
+    """Return Slack's application-level error code, if present."""
+    return str(get_slack_api_error_data(ex).get("error") or "")
+
+
+def get_slack_api_status_code(ex: SlackApiError) -> int | None:
+    """Return the HTTP status carried by a Slack API error, if present."""
+    return getattr(getattr(ex, "response", None), "status_code", None)
+
+
+def is_transient_slack_api_error(ex: SlackApiError, error_code: str) -> bool:
+    """Return whether Slack reported a retryable API or HTTP condition."""
+    status_code = get_slack_api_status_code(ex)
+    return bool(
+        status_code in {408, 429}
+        or (status_code is not None and 500 <= status_code < 600)
+        or error_code in _TRANSIENT_SLACK_API_ERROR_CODES
+    )
+
+
+def is_transient_slack_transport_error(ex: Exception) -> bool:
+    """Classify raw Slack WebClient transport and external-upload failures."""
+    if isinstance(ex, HTTPError):
+        return ex.code in {408, 429} or 500 <= ex.code < 600
+    return isinstance(ex, SLACK_TRANSIENT_TRANSPORT_ERRORS)
+
+
+def is_retryable_slack_transport_error(ex: Exception) -> bool:
+    """Return whether an application retry cannot duplicate an accepted write."""
+    if isinstance(ex, HTTPError):
+        return is_transient_slack_transport_error(ex)
+    return isinstance(ex, SlackClientNotConnectedError)
+
+
+def get_slack_client(*, for_delivery: bool = False) -> WebClient:
+    """Build a Slack client without nested SDK retries for delivery writes."""
     token: str = app.config["SLACK_API_TOKEN"]
     if callable(token):
         token = token()
@@ -84,11 +211,15 @@ def get_slack_client() -> WebClient:
         token=token,
         proxy=app.config["SLACK_PROXY"],
         timeout=app.config["SLACK_API_TIMEOUT"],
+        retry_handlers=[] if for_delivery else None,
     )
 
     max_retry_count = app.config.get("SLACK_API_RATE_LIMIT_RETRY_COUNT", 2)
-    rate_limit_handler = RateLimitErrorRetryHandler(max_retry_count=max_retry_count)
-    client.retry_handlers.append(rate_limit_handler)
+    if not for_delivery:
+        rate_limit_handler = RateLimitErrorRetryHandler(
+            max_retry_count=max_retry_count,
+        )
+        client.retry_handlers.append(rate_limit_handler)
 
     logger.debug("Slack client configured with %d rate limit retries", max_retry_count)
 
@@ -112,9 +243,23 @@ def get_team_id() -> Optional[str]:
     return team_id or None
 
 
+def _get_slack_channels_cache_key(team_id: Optional[str]) -> str:
+    cache_key = "slack_conversations_list"
+    return f"{cache_key}_{team_id}" if team_id else cache_key
+
+
+def _slack_channel_cache_uses_report_session() -> bool:
+    """Return whether cache writes commit or roll back the report DB session."""
+    return isinstance(cache_manager.cache.cache, SupersetMetastoreCache)
+
+
 def get_channels(
-    team_id: Optional[str] = None, **kwargs: Any
-) -> list[SlackChannelSchema]:
+    team_id: Optional[str] = None,
+    *,
+    force: bool = False,
+    cache_timeout: int | None = None,
+    cache: bool = True,
+) -> list[SlackChannel]:
     """
     Retrieves a list of all conversations accessible by the bot
     from the Slack API, and caches results (to avoid rate limits).
@@ -128,27 +273,84 @@ def get_channels(
         distinct workspaces never share cached channel lists; when unset, the
         legacy cache key is used so that upgrading does not invalidate existing
         caches.
-    :param kwargs: forwarded to the memoized fetch (``force``, ``cache_timeout``,
-        ``cache``).
+    Cache reads and writes are best-effort; a backend failure does not replace
+    successfully fetched Slack data.
     """
     if team_id is None:
         team_id = get_team_id()
-    cache_key = "slack_conversations_list"
-    if team_id:
-        cache_key = f"{cache_key}_{team_id}"
-    return _get_channels(cache_key, team_id=team_id, **kwargs)
+    channels, _ = _get_channels_safely(
+        team_id=team_id,
+        force=force,
+        cache=cache,
+        cache_timeout=cache_timeout,
+        write_metastore_cache=True,
+    )
+    return channels
+
+
+def _get_channels_safely(
+    *,
+    team_id: str | None,
+    force: bool,
+    cache: bool,
+    write_metastore_cache: bool,
+    cache_timeout: int | None = None,
+) -> tuple[list[SlackChannel], bool]:
+    """Fetch channels with best-effort cache access and hit provenance."""
+    cache_key = _get_slack_channels_cache_key(team_id)
+    effective_timeout = (
+        app.config["SLACK_CACHE_TIMEOUT"] if cache_timeout is None else cache_timeout
+    )
+    cache_enabled = cache and effective_timeout != CACHE_DISABLED_TIMEOUT
+
+    if cache_enabled and not force:
+        try:
+            cached_channels = cache_manager.cache.get(cache_key)
+        except Exception:  # pylint: disable=broad-exception-caught
+            cached_channels = None
+            logger.warning(
+                "Could not read cached Slack channels; fetching from Slack",
+                exc_info=True,
+            )
+        if cached_channels is not None:
+            return cached_channels, True
+
+    channels = _get_channels(cache_key, team_id=team_id, cache=False)
+    if cache_enabled and (
+        write_metastore_cache or not _slack_channel_cache_uses_report_session()
+    ):
+        try:
+            cache_manager.cache.set(
+                cache_key,
+                channels,
+                timeout=effective_timeout,
+            )
+        except Exception:  # pylint: disable=broad-exception-caught
+            logger.warning(
+                "Could not cache Slack channels",
+                exc_info=True,
+            )
+    return channels, False
+
+
+def _get_channels_with_cache_status() -> tuple[list[SlackChannel], bool]:
+    """Fetch channels and cache-hit provenance using one cache read."""
+    return _get_channels_safely(
+        team_id=get_team_id(),
+        force=False,
+        cache=True,
+        write_metastore_cache=False,
+    )
 
 
 @cache_util.memoized_func(
     key="{cache_key}",
     cache=cache_manager.cache,
 )
-def _get_channels(
-    cache_key: str, team_id: Optional[str] = None
-) -> list[SlackChannelSchema]:
+def _get_channels(cache_key: str, team_id: Optional[str] = None) -> list[SlackChannel]:
     client = get_slack_client()
     channel_schema = SlackChannelSchema()
-    channels: list[SlackChannelSchema] = []
+    channels: list[SlackChannel] = []
     extra_params = {"types": _SLACK_CONVERSATION_TYPES}
     if team_id:
         extra_params["team_id"] = team_id
@@ -194,37 +396,16 @@ def _get_channels(
         raise
 
 
-def get_channels_with_search(
-    search_string: str = "",
-    types: Optional[list[SlackChannelTypes]] = None,
-    exact_match: bool = False,
-    force: bool = False,
-) -> list[SlackChannelSchema]:
-    """
-    The slack api is paginated but does not include search, so we need to fetch
-    all channels and filter them ourselves
-    This will search by slack name or id
-    """
-    try:
-        channels = get_channels(
-            force=force,
-            cache_timeout=app.config["SLACK_CACHE_TIMEOUT"],
-        )
-    except SlackApiError as ex:
-        # Check if it's a rate limit error
-        status_code = getattr(ex.response, "status_code", None)
-        if status_code == 429:
-            raise SupersetException(
-                f"Slack API rate limit exceeded: {ex}. "
-                "For large workspaces, consider increasing "
-                "SLACK_API_RATE_LIMIT_RETRY_COUNT"
-            ) from ex
-        raise SupersetException(f"Failed to list channels: {ex}") from ex
-    except SlackClientError as ex:
-        raise SupersetException(f"Failed to list channels: {ex}") from ex
-
-    if types and not len(types) == len(SlackChannelTypes):
-        conditions: list[Callable[[SlackChannelSchema], bool]] = []
+def _filter_slack_channels(
+    channels: list[SlackChannel],
+    *,
+    search_string: str,
+    types: Optional[list[SlackChannelTypes]],
+    exact_match: bool,
+) -> list[SlackChannel]:
+    """Filter a complete Slack channel listing by type and target."""
+    if types and len(types) != len(SlackChannelTypes):
+        conditions: list[Callable[[SlackChannel], bool]] = []
         if SlackChannelTypes.PUBLIC in types:
             conditions.append(lambda channel: not channel["is_private"])
         if SlackChannelTypes.PRIVATE in types:
@@ -234,25 +415,193 @@ def get_channels_with_search(
             channel for channel in channels if any(cond(channel) for cond in conditions)
         ]
 
-    # The search string can be multiple channels separated by commas
-    if search_string:
-        search_array = recipients_string_to_list(search_string)
-        channels = [
-            channel
-            for channel in channels
-            if any(
-                (
-                    search.lower() == channel["name"].lower()
-                    or search.lower() == channel["id"].lower()
-                    if exact_match
-                    else (
-                        search.lower() in channel["name"].lower()
-                        or search.lower() in channel["id"].lower()
-                    )
+    if not search_string:
+        return channels
+
+    search_array = recipients_string_to_list(search_string)
+    return [
+        channel
+        for channel in channels
+        if any(
+            (
+                search.casefold() == channel["name"].casefold()
+                or search.casefold() == channel["id"].casefold()
+                if exact_match
+                else (
+                    search.casefold() in channel["name"].casefold()
+                    or search.casefold() in channel["id"].casefold()
                 )
-                for search in search_array
             )
-        ]
+            for search in search_array
+        )
+    ]
+
+
+@overload
+def get_channels_with_search(
+    search_string: str = "",
+    types: Optional[list[SlackChannelTypes]] = None,
+    exact_match: bool = False,
+    force: bool = False,
+    cache: bool = True,
+    *,
+    return_cache_status: Literal[False] = False,
+) -> list[SlackChannel]: ...
+
+
+@overload
+def get_channels_with_search(
+    search_string: str = "",
+    types: Optional[list[SlackChannelTypes]] = None,
+    exact_match: bool = False,
+    force: bool = False,
+    cache: bool = True,
+    *,
+    return_cache_status: Literal[True],
+) -> tuple[list[SlackChannel], bool]: ...
+
+
+def get_channels_with_search(
+    search_string: str = "",
+    types: Optional[list[SlackChannelTypes]] = None,
+    exact_match: bool = False,
+    force: bool = False,
+    cache: bool = True,
+    *,
+    return_cache_status: bool = False,
+) -> list[SlackChannel] | tuple[list[SlackChannel], bool]:
+    """
+    The slack api is paginated but does not include search, so we need to fetch
+    all channels and filter them ourselves
+    This will search by slack name or id
+    """
+    used_cache = False
+    cache_timeout = app.config["SLACK_CACHE_TIMEOUT"]
+    cache_enabled = cache and cache_timeout != CACHE_DISABLED_TIMEOUT
+    try:
+        if return_cache_status and cache_enabled and not force:
+            channels, used_cache = _get_channels_with_cache_status()
+        else:
+            channels = get_channels(
+                force=force,
+                cache=cache_enabled,
+                cache_timeout=cache_timeout,
+            )
+    except SlackApiError as ex:
+        error_code = get_slack_api_error_code(ex)
+        error_class = (
+            SlackChannelListingError
+            if is_transient_slack_api_error(ex, error_code)
+            else SlackChannelListingClientError
+        )
+        raise error_class(f"Failed to list channels: {ex}") from ex
+    except SLACK_TRANSIENT_TRANSPORT_ERRORS as ex:
+        error_class = (
+            SlackChannelListingError
+            if is_transient_slack_transport_error(ex)
+            else SlackChannelListingClientError
+        )
+        raise error_class(f"Failed to list channels: {ex}") from ex
+    except (SlackSDKClientError, SlackClientError) as ex:
+        raise SlackChannelListingClientError(f"Failed to list channels: {ex}") from ex
+
+    channels = _filter_slack_channels(
+        channels,
+        search_string=search_string,
+        types=types,
+        exact_match=exact_match,
+    )
+    return (channels, used_cache) if return_cache_status else channels
+
+
+def refresh_cached_slack_channels_with_search(
+    search_string: str = "",
+    types: Optional[list[SlackChannelTypes]] = None,
+    exact_match: bool = False,
+) -> list[SlackChannel]:
+    """Refresh stale channels with a best-effort cache-backend cooldown.
+
+    External cache backends record a cooldown only after the refreshed listing
+    is stored successfully. Disabled and metastore-backed caches use an uncached
+    request without a cooldown because metastore writes commit the report
+    transaction. Concurrent workers can still refresh in parallel when the
+    backend cannot provide transaction-safe coordination.
+    """
+    team_id = get_team_id()
+    cache_key = _get_slack_channels_cache_key(team_id)
+    cooldown_key = f"{cache_key}_refresh_cooldown"
+    cache_timeout = app.config["SLACK_CACHE_TIMEOUT"]
+
+    if (
+        _slack_channel_cache_uses_report_session()
+        or cache_timeout == CACHE_DISABLED_TIMEOUT
+    ):
+        return get_channels_with_search(
+            search_string=search_string,
+            types=types,
+            exact_match=exact_match,
+            force=True,
+            cache=False,
+        )
+
+    try:
+        refresh_is_recent = cache_manager.cache.get(cooldown_key) is not None
+    except Exception:  # pylint: disable=broad-exception-caught
+        refresh_is_recent = False
+        logger.warning(
+            "Could not read Slack channel refresh cooldown; refreshing from Slack",
+            exc_info=True,
+        )
+
+    if refresh_is_recent:
+        channels, _ = _get_channels_with_cache_status()
+        return _filter_slack_channels(
+            channels,
+            search_string=search_string,
+            types=types,
+            exact_match=exact_match,
+        )
+
+    refreshed_channels = get_channels_with_search(
+        force=True,
+        cache=False,
+    )
+    try:
+        cache_updated = (
+            cache_manager.cache.set(
+                cache_key,
+                refreshed_channels,
+                timeout=cache_timeout,
+            )
+            is not False
+        )
+    except Exception:  # pylint: disable=broad-exception-caught
+        cache_updated = False
+        logger.warning(
+            "Could not cache refreshed Slack channels",
+            exc_info=True,
+        )
+
+    channels = _filter_slack_channels(
+        refreshed_channels,
+        search_string=search_string,
+        types=types,
+        exact_match=exact_match,
+    )
+    if not cache_updated:
+        return channels
+
+    try:
+        cache_manager.cache.set(
+            cooldown_key,
+            True,
+            timeout=_SLACK_CHANNEL_REFRESH_COOLDOWN_SECONDS,
+        )
+    except Exception:  # pylint: disable=broad-exception-caught
+        logger.warning(
+            "Could not record Slack channel refresh cooldown",
+            exc_info=True,
+        )
     return channels
 
 
@@ -261,7 +610,7 @@ _SCOPE_MISSING_ERROR_CODES = frozenset(
 )
 
 
-def should_use_v2_api() -> bool:
+def should_use_v2_api(*, raise_on_error: bool = False) -> bool:
     if not feature_flag_manager.is_feature_enabled("ALERT_REPORT_SLACK_V2"):
         _emit_v1_flag_off_deprecation()
         return False
@@ -282,17 +631,11 @@ def should_use_v2_api() -> bool:
         # Only the scope-missing branch is a v1-deprecation signal; other
         # SlackApiError codes (invalid_auth, ratelimited, server errors, etc.)
         # are unrelated probe failures and should not be reported as a missing
-        # scope. We still fall back to v1 in both cases so a transient probe
-        # failure doesn't break sends — operators get an actionable log either
-        # way.
-        # `response` is normally a SlackResponse whose payload lives in `.data`,
-        # but the SDK (and our tests) can also hand back a plain dict. Read the
-        # error code in either shape so the scope-missing branch isn't missed.
-        response = getattr(ex, "response", None)
-        data = getattr(response, "data", None)
-        if not isinstance(data, dict):
-            data = response if isinstance(response, dict) else {}
-        error_code = data.get("error", "")
+        # scope. Scope errors continue through v1 for compatibility. Other
+        # failures also fall back for text-only reports, while file-bearing
+        # reports request an exception so monitoring retains the system/client
+        # classification.
+        error_code = get_slack_api_error_code(ex)
         if error_code in _SCOPE_MISSING_ERROR_CODES:
             # The DeprecationWarning fires once per process, but the actionable
             # log line fires every send so operators see it in their report logs.
@@ -305,27 +648,49 @@ def should_use_v2_api() -> bool:
             )
         else:
             logger.warning(
-                "Slack v2 probe failed with error %r; falling back to the "
-                "deprecated v1 API for this send. Investigate the underlying "
-                "Slack API error — this is not a missing-scope problem.",
+                "Slack v2 probe failed with error %r. Investigate the underlying "
+                "Slack API error; this is not a missing-scope problem.",
                 error_code or str(ex),
             )
+            if raise_on_error:
+                error_class = (
+                    SlackV2ProbeError
+                    if is_transient_slack_api_error(ex, error_code)
+                    else SlackV2ProbeClientError
+                )
+                raise error_class(
+                    f"Slack v2 availability probe failed: {error_code or str(ex)}"
+                ) from ex
         return False
-    except SlackSDKClientError as ex:
-        # Non-API SDK failures (e.g. SlackClientNotConnectedError,
-        # SlackRequestError, SlackClientConfigurationError) are not subclasses
-        # of SlackApiError, so without this branch they would escape the probe
-        # raw. The caller runs this probe *before* the mapped Slack send `try`,
-        # so an un-caught probe error aborts the entire recipient loop instead
-        # of failing a single recipient. Treat any probe connection/transport
-        # failure as "v2 unavailable" and fall back to the deprecated v1 API,
-        # matching the SlackApiError behavior above.
+    except SLACK_TRANSIENT_TRANSPORT_ERRORS as ex:
         logger.warning(
-            "Slack v2 probe failed to connect (%s: %s); falling back to the "
-            "deprecated v1 API for this send.",
+            "Slack v2 probe failed (%s: %s).",
             type(ex).__name__,
             ex,
         )
+        if raise_on_error:
+            error_class = (
+                SlackV2ProbeError
+                if is_transient_slack_transport_error(ex)
+                else SlackV2ProbeClientError
+            )
+            raise error_class(
+                f"Slack v2 availability probe failed: {type(ex).__name__}: {ex}"
+            ) from ex
+        return False
+    except SlackSDKClientError as ex:
+        # Permanent SDK request/configuration failures are operator-fixable.
+        # Text reports retain v1 compatibility, while file-bearing reports ask
+        # the command to preserve the client-error classification.
+        logger.warning(
+            "Slack v2 probe failed (%s: %s).",
+            type(ex).__name__,
+            ex,
+        )
+        if raise_on_error:
+            raise SlackV2ProbeClientError(
+                f"Slack v2 availability probe failed: {type(ex).__name__}: {ex}"
+            ) from ex
         return False
 
 

tests/integration_tests/reports/commands_tests.py

@@ -18,9 +18,10 @@ from collections.abc import Iterator
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Optional
-from unittest.mock import call, Mock, patch
+from unittest.mock import ANY, call, Mock, patch
 from uuid import UUID, uuid4
 
+import pandas as pd
 import pytest
 from flask.ctx import AppContext
 from flask_appbuilder.security.sqla.models import User
@@ -65,6 +66,7 @@ from superset.commands.report.execute import (
     BaseReportState,
 )
 from superset.commands.report.log_prune import AsyncPruneReportScheduleLogCommand
+from superset.daos.report import ReportScheduleDAO
 from superset.exceptions import SupersetException
 from superset.key_value.models import KeyValueEntry
 from superset.models.core import Database
@@ -73,6 +75,7 @@ from superset.models.slice import Slice
 from superset.reports.models import (
     ReportDataFormat,
     ReportExecutionLog,
+    ReportRecipients,
     ReportRecipientType,
     ReportSchedule,
     ReportScheduleType,
@@ -115,6 +118,20 @@ pytestmark = pytest.mark.usefixtures(
 )
 
 
+def _configure_v2_upload_client(client: Mock) -> Mock:
+    """Configure a Slack SDK-shaped three-phase file upload mock."""
+    client.timeout = 30
+    client.proxy = None
+    client.ssl = None
+    client.files_getUploadURLExternal.return_value = {
+        "file_id": "F1",
+        "upload_url": "https://files.slack.com/upload/F1",
+    }
+    client._upload_file.return_value = Mock(status=200, body="ok")
+    client.files_completeUploadExternal.return_value = {"files": [{"id": "F1"}]}
+    return client
+
+
 def get_target_from_report_schedule(report_schedule: ReportSchedule) -> list[str]:
     return [
         json.loads(recipient.recipient_config_json)["target"]
@@ -1463,7 +1480,7 @@ def test_email_dashboard_report_schedule_force_screenshot(
 
 
 @pytest.mark.usefixtures("create_report_slack_chart")
-@patch("superset.commands.report.execute.get_channels_with_search")
+@patch("superset.reports.notifications.slack_channel_resolver.get_channels_with_search")
 @patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
 @patch("superset.reports.notifications.slackv2.get_slack_client")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
@@ -1480,15 +1497,19 @@ def test_slack_chart_report_schedule_converts_to_v2(
     """
     # setup screenshot mock
     screenshot_mock.return_value = SCREENSHOT_FILE
+    slack_client = _configure_v2_upload_client(slack_client_mock.return_value)
     channel_id = "slack_channel_id"
-    get_channels_with_search_mock.return_value = [
-        {
-            "id": channel_id,
-            "name": "slack_channel",
-            "is_member": True,
-            "is_private": False,
-        },
-    ]
+    get_channels_with_search_mock.return_value = (
+        [
+            {
+                "id": channel_id,
+                "name": "slack_channel",
+                "is_member": True,
+                "is_private": False,
+            },
+        ],
+        False,
+    )
 
     with freeze_time("2020-01-01T00:00:00Z"):
         with patch(
@@ -1499,13 +1520,10 @@ def test_slack_chart_report_schedule_converts_to_v2(
             ).run()
 
             assert (
-                slack_client_mock.return_value.files_upload_v2.call_args[1]["channel"]
+                slack_client.files_completeUploadExternal.call_args[1]["channel_id"]
                 == channel_id
             )
-            assert (
-                slack_client_mock.return_value.files_upload_v2.call_args[1]["file"]
-                == SCREENSHOT_FILE
-            )
+            assert slack_client._upload_file.call_args[1]["data"] == SCREENSHOT_FILE
 
             # Assert that the report recipients were updated
             assert create_report_slack_chart.recipients[
@@ -1518,14 +1536,10 @@ def test_slack_chart_report_schedule_converts_to_v2(
 
             # Assert logs are correct
             assert_log(ReportState.SUCCESS)
-            # this will send a warning
-            assert statsd_mock.call_args_list[0] == call(
-                "reports.slack.send.warning", 1
-            )
-            assert statsd_mock.call_args_list[1] == call("reports.slack.send.ok", 1)
+            statsd_mock.assert_called_once_with("reports.slack.send.ok", 1)
 
 
-@patch("superset.commands.report.execute.get_channels_with_search")
+@patch("superset.reports.notifications.slack_channel_resolver.get_channels_with_search")
 @patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
 @patch("superset.reports.notifications.slackv2.get_slack_client")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
@@ -1541,19 +1555,23 @@ def test_slack_chart_report_schedule_converts_to_v2_channel_with_hash(
     """
     # setup screenshot mock
     screenshot_mock.return_value = SCREENSHOT_FILE
+    slack_client = _configure_v2_upload_client(slack_client_mock.return_value)
     channel_id = "slack_channel_id"
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
         slack_channel="#slack_channel", chart=chart
     )
-    get_channels_with_search_mock.return_value = [
-        {
-            "id": channel_id,
-            "name": "slack_channel",
-            "is_member": True,
-            "is_private": False,
-        },
-    ]
+    get_channels_with_search_mock.return_value = (
+        [
+            {
+                "id": channel_id,
+                "name": "slack_channel",
+                "is_member": True,
+                "is_private": False,
+            },
+        ],
+        False,
+    )
 
     with freeze_time("2020-01-01T00:00:00Z"):
         with patch(
@@ -1564,13 +1582,10 @@ def test_slack_chart_report_schedule_converts_to_v2_channel_with_hash(
             ).run()
 
             assert (
-                slack_client_mock.return_value.files_upload_v2.call_args[1]["channel"]
+                slack_client.files_completeUploadExternal.call_args[1]["channel_id"]
                 == channel_id
             )
-            assert (
-                slack_client_mock.return_value.files_upload_v2.call_args[1]["file"]
-                == SCREENSHOT_FILE
-            )
+            assert slack_client._upload_file.call_args[1]["data"] == SCREENSHOT_FILE
 
             # Assert that the report recipients were updated
             assert report_schedule.recipients[0].recipient_config_json == json.dumps(
@@ -1580,28 +1595,22 @@ def test_slack_chart_report_schedule_converts_to_v2_channel_with_hash(
 
             # Assert logs are correct
             assert_log(ReportState.SUCCESS)
-            # this will send a warning
-            assert statsd_mock.call_args_list[0] == call(
-                "reports.slack.send.warning", 1
-            )
-            assert statsd_mock.call_args_list[1] == call("reports.slack.send.ok", 1)
+            statsd_mock.assert_called_once_with("reports.slack.send.ok", 1)
 
     cleanup_report_schedule(report_schedule)
 
 
-@patch("superset.commands.report.execute.get_channels_with_search")
+@patch("superset.reports.notifications.slack_channel_resolver.get_channels_with_search")
 @patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
 @patch("superset.reports.notifications.slackv2.get_slack_client")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
-def test_slack_chart_report_schedule_fails_to_converts_to_v2(
+def test_slack_chart_report_schedule_failed_v2_conversion_rejects_v1_file_upload(
     screenshot_mock,
     slack_client_mock,
     slack_should_use_v2_api_mock,
     get_channels_with_search_mock,
 ):
-    """
-    ExecuteReport Command: Test converting a Slack report to v2 fails.
-    """
+    """A failed Slack v2 conversion rejects unsupported Slack v1 file uploads."""
     # setup screenshot mock
     screenshot_mock.return_value = SCREENSHOT_FILE
     channel_id = "slack_channel_id"
@@ -1609,34 +1618,50 @@ def test_slack_chart_report_schedule_fails_to_converts_to_v2(
     report_schedule = create_report_notification(
         slack_channel="#slack_channel,my_member_ID", chart=chart
     )
-    get_channels_with_search_mock.return_value = [
-        {
-            "id": channel_id,
-            "name": "slack_channel",
-            "is_member": True,
-            "is_private": False,
-        },
-    ]
-
-    with pytest.raises(ReportScheduleSystemErrorsException):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, report_schedule.id, datetime.utcnow()
-        ).run()
-
-    # Assert failuer with proper log
-    expected_message = (
-        "Failed to update slack recipients to v2: "
-        "Could not find the following channels: my_member_ID"
+    get_channels_with_search_mock.return_value = (
+        [
+            {
+                "id": channel_id,
+                "name": "slack_channel",
+                "is_member": True,
+                "is_private": False,
+            },
+        ],
+        False,
     )
-    assert_log(ReportState.ERROR, error_message=expected_message)
 
-    # Assert that previous configuration was kept for manual correction
-    assert report_schedule.recipients[0].recipient_config_json == json.dumps(
-        {"target": "#slack_channel,my_member_ID"}
-    )
-    assert report_schedule.recipients[0].type == ReportRecipientType.SLACK
+    try:
+        with (
+            patch(
+                "superset.extensions.stats_logger_manager.instance.gauge"
+            ) as statsd_mock,
+            pytest.raises(ReportScheduleClientErrorsException),
+        ):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, report_schedule.id, datetime.utcnow()
+            ).run()
 
-    cleanup_report_schedule(report_schedule)
+        expected_message = (
+            "Slack v1 file uploads are no longer supported because Slack retired "
+            "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+            "both the `channels:read` and `groups:read` scopes so the recipient can "
+            "be upgraded to Slack v2. Slack v2 upgrade failed: Failed to update "
+            "slack recipients to v2: Could not find the following channels: "
+            "my_member_ID"
+        )
+        assert_log(ReportState.ERROR, error_message=expected_message)
+
+        # Keep the previous configuration for manual correction.
+        assert report_schedule.recipients[0].recipient_config_json == json.dumps(
+            {"target": "#slack_channel,my_member_ID"}
+        )
+        assert report_schedule.recipients[0].type == ReportRecipientType.SLACK
+        slack_client_mock.assert_not_called()
+        assert (
+            statsd_mock.call_args_list.count(call("reports.slack.send.warning", 1)) == 1
+        )
+    finally:
+        cleanup_report_schedule(report_schedule)
 
 
 @pytest.mark.usefixtures("create_report_slack_chartv2")
@@ -1654,6 +1679,7 @@ def test_slack_chart_report_schedule_v2(
     """
     # setup screenshot mock
     screenshot_mock.return_value = SCREENSHOT_FILE
+    slack_client = _configure_v2_upload_client(slack_client_mock.return_value)
 
     with freeze_time("2020-01-01T00:00:00Z"):
         with patch(
@@ -1664,13 +1690,10 @@ def test_slack_chart_report_schedule_v2(
             ).run()
 
             assert (
-                slack_client_mock.return_value.files_upload_v2.call_args[1]["channel"]
+                slack_client.files_completeUploadExternal.call_args[1]["channel_id"]
                 == "slack_channel_id"
             )
-            assert (
-                slack_client_mock.return_value.files_upload_v2.call_args[1]["file"]
-                == SCREENSHOT_FILE
-            )
+            assert slack_client._upload_file.call_args[1]["data"] == SCREENSHOT_FILE
 
             # Assert logs are correct
             assert_log(ReportState.SUCCESS)
@@ -1747,9 +1770,7 @@ def test_slack_chart_report_schedule_with_csv(
     slack_should_use_v2_api_mock,
     create_report_slack_chart_with_csv,
 ):
-    """
-    ExecuteReport Command: Test chart slack report V1 schedule with CSV
-    """
+    """A v1 CSV report fails before calling Slack's retired upload API."""
     # setup csv mock
     response = Mock()
     mock_open.return_value = response
@@ -1757,28 +1778,20 @@ def test_slack_chart_report_schedule_with_csv(
     mock_urlopen.return_value.getcode.return_value = 200
     response.read.return_value = CSV_FILE
 
-    notification_targets = get_target_from_report_schedule(
-        create_report_slack_chart_with_csv
-    )
-
-    channel_name = notification_targets[0]
-
     with freeze_time("2020-01-01T00:00:00Z"):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, create_report_slack_chart_with_csv.id, datetime.utcnow()
-        ).run()
+        with pytest.raises(ReportScheduleClientErrorsException):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, create_report_slack_chart_with_csv.id, datetime.utcnow()
+            ).run()
 
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["channels"]
-            == channel_name
+        expected_message = (
+            "Slack v1 file uploads are no longer supported because Slack retired "
+            "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+            "both the `channels:read` and `groups:read` scopes so the recipient can "
+            "be upgraded to Slack v2."
         )
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["file"]
-            == CSV_FILE
-        )
-
-        # Assert logs are correct
-        assert_log(ReportState.SUCCESS)
+        assert_log(ReportState.ERROR, error_message=expected_message)
+        slack_client_mock_class.assert_not_called()
 
 
 @pytest.mark.usefixtures(
@@ -1797,9 +1810,7 @@ def test_slack_chart_report_schedule_with_xlsx(
     slack_should_use_v2_api_mock: Mock,
     create_report_slack_chart_with_xlsx: ReportSchedule,
 ) -> None:
-    """
-    ExecuteReport Command: Test chart slack report V1 schedule with Excel
-    """
+    """A v1 XLSX report fails before calling Slack's retired upload API."""
     # setup xlsx mock
     response = Mock()
     mock_open.return_value = response
@@ -1807,28 +1818,20 @@ def test_slack_chart_report_schedule_with_xlsx(
     mock_urlopen.return_value.getcode.return_value = 200
     response.read.return_value = XLSX_FILE
 
-    notification_targets = get_target_from_report_schedule(
-        create_report_slack_chart_with_xlsx
-    )
-
-    channel_name = notification_targets[0]
-
     with freeze_time("2020-01-01T00:00:00Z"):
-        AsyncExecuteReportScheduleCommand(
-            TEST_ID, create_report_slack_chart_with_xlsx.id, datetime.utcnow()
-        ).run()
+        with pytest.raises(ReportScheduleClientErrorsException):
+            AsyncExecuteReportScheduleCommand(
+                TEST_ID, create_report_slack_chart_with_xlsx.id, datetime.utcnow()
+            ).run()
 
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["channels"]
-            == channel_name
+        expected_message = (
+            "Slack v1 file uploads are no longer supported because Slack retired "
+            "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+            "both the `channels:read` and `groups:read` scopes so the recipient can "
+            "be upgraded to Slack v2."
         )
-        assert (
-            slack_client_mock_class.return_value.files_upload.call_args[1]["file"]
-            == XLSX_FILE
-        )
-
-        # Assert logs are correct
-        assert_log(ReportState.SUCCESS)
+        assert_log(ReportState.ERROR, error_message=expected_message)
+        slack_client_mock_class.assert_not_called()
 
 
 @pytest.mark.usefixtures(
@@ -1898,6 +1901,209 @@ def test_slack_chart_report_schedule_with_text(
         assert_log(ReportState.SUCCESS)
 
 
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_text"
+)
+@patch(
+    "superset.reports.notifications.slack_channel_resolver.get_channels_with_search",
+    return_value=([], False),
+)
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
+@patch("superset.reports.notifications.slack.get_slack_client")
+@patch("superset.commands.report.execute.get_chart_dataframe")
+def test_slack_text_fallback_persists_success_for_multiple_recipient_rows(
+    dataframe_mock,
+    slack_client_mock,
+    slack_should_use_v2_api_mock,
+    get_channels_with_search_mock,
+    create_report_slack_chart_with_text,
+):
+    """Failed migration sends every text recipient and persists v1 success."""
+    dataframe_mock.return_value = pd.DataFrame({"value": [1]})
+    original_configs = [
+        json.dumps({"target": "private-a"}),
+        json.dumps({"target": "private-b"}),
+    ]
+    create_report_slack_chart_with_text.recipients[
+        0
+    ].recipient_config_json = original_configs[0]
+    create_report_slack_chart_with_text.recipients.append(
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=original_configs[1],
+        )
+    )
+    db.session.commit()
+    report_schedule_id = create_report_slack_chart_with_text.id
+
+    with patch(
+        "superset.extensions.stats_logger_manager.instance.gauge"
+    ) as statsd_mock:
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID,
+            report_schedule_id,
+            datetime.utcnow(),
+        ).run()
+
+    db.session.expire_all()
+    persisted_schedule = db.session.get(ReportSchedule, report_schedule_id)
+    assert persisted_schedule is not None
+    assert persisted_schedule.last_state == ReportState.SUCCESS
+    assert all(
+        recipient.type == ReportRecipientType.SLACK
+        for recipient in persisted_schedule.recipients
+    )
+    assert {
+        recipient.recipient_config_json for recipient in persisted_schedule.recipients
+    } == set(original_configs)
+    assert {
+        slack_call.kwargs["channel"]
+        for slack_call in slack_client_mock.return_value.chat_postMessage.call_args_list
+    } == {"private-a", "private-b"}
+    assert slack_client_mock.return_value.chat_postMessage.call_count == 2
+    assert statsd_mock.call_args_list == [
+        call("reports.slack.send.ok", 1),
+        call("reports.slack.send.ok", 1),
+    ]
+    assert slack_should_use_v2_api_mock.call_count == 1
+    get_channels_with_search_mock.assert_called_once_with(
+        search_string=ANY,
+        types=ANY,
+        exact_match=True,
+        force=False,
+        return_cache_status=True,
+    )
+    success_logs = (
+        db.session.query(ReportExecutionLog)
+        .filter(
+            ReportExecutionLog.report_schedule_id == report_schedule_id,
+            ReportExecutionLog.state == ReportState.SUCCESS,
+        )
+        .all()
+    )
+    assert len(success_logs) == 1
+    assert "deprecated Slack v1" in success_logs[0].error_message
+
+
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_text"
+)
+@patch(
+    "superset.reports.notifications.slack_channel_resolver.get_channels_with_search",
+    return_value=([], False),
+)
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
+@patch("superset.reports.notifications.slack.get_slack_client")
+@patch("superset.commands.report.execute.get_chart_dataframe")
+@patch("superset.reports.notifications.email.send_email_smtp")
+@pytest.mark.parametrize(
+    "error_notification_fails",
+    [False, True],
+    ids=["notification-succeeds", "notification-fails"],
+)
+def test_slack_text_fallback_persists_later_recipient_ambiguous_failure(
+    email_mock,
+    dataframe_mock,
+    slack_client_mock,
+    slack_should_use_v2_api_mock,
+    get_channels_with_search_mock,
+    create_report_slack_chart_with_text,
+    error_notification_fails,
+):
+    """Mixed-outcome errors persist the fallback warning exactly once."""
+    dataframe_mock.return_value = pd.DataFrame({"value": [1]})
+    original_configs = [
+        json.dumps({"target": "private-a"}),
+        json.dumps({"target": "private-b"}),
+    ]
+    create_report_slack_chart_with_text.recipients[
+        0
+    ].recipient_config_json = original_configs[0]
+    create_report_slack_chart_with_text.recipients.append(
+        ReportRecipients(
+            type=ReportRecipientType.SLACK,
+            recipient_config_json=original_configs[1],
+        )
+    )
+    db.session.commit()
+    report_schedule_id = create_report_slack_chart_with_text.id
+
+    successful_channels: list[str] = []
+
+    def chat_side_effect(channel, text):
+        if not successful_channels:
+            successful_channels.append(channel)
+            return {"ok": True}
+        if channel != successful_channels[0]:
+            raise SlackApiError(
+                message="service unavailable",
+                response={"ok": False, "error": "service_unavailable"},
+            )
+        return {"ok": True}
+
+    slack_client_mock.return_value.chat_postMessage.side_effect = chat_side_effect
+    if error_notification_fails:
+        email_mock.side_effect = RuntimeError("SMTP unavailable")
+
+    with (
+        patch("time.sleep"),
+        pytest.raises(ReportScheduleSystemErrorsException),
+    ):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID,
+            report_schedule_id,
+            datetime.utcnow(),
+        ).run()
+
+    db.session.expire_all()
+    persisted_schedule = db.session.get(ReportSchedule, report_schedule_id)
+    assert persisted_schedule is not None
+    assert persisted_schedule.last_state == ReportState.ERROR
+    assert all(
+        recipient.type == ReportRecipientType.SLACK
+        for recipient in persisted_schedule.recipients
+    )
+    assert {
+        recipient.recipient_config_json for recipient in persisted_schedule.recipients
+    } == set(original_configs)
+    call_channels = [
+        slack_call.kwargs["channel"]
+        for slack_call in slack_client_mock.return_value.chat_postMessage.call_args_list
+    ]
+    successful_channel = call_channels[0]
+    failed_channel = ({"private-a", "private-b"} - {successful_channel}).pop()
+    assert call_channels == [successful_channel, failed_channel]
+    execution_logs = (
+        db.session.query(ReportExecutionLog)
+        .filter(ReportExecutionLog.report_schedule_id == report_schedule_id)
+        .all()
+    )
+    log_states = {log.state for log in execution_logs}
+    assert log_states == {ReportState.ERROR}
+    error_messages = [
+        log.error_message or ""
+        for log in execution_logs
+        if log.state == ReportState.ERROR
+    ]
+    warning_messages = [
+        message for message in error_messages if "deprecated Slack v1" in message
+    ]
+    assert len(warning_messages) == 1
+    assert warning_messages[0].count("deprecated Slack v1") == 1
+    assert "service unavailable" in warning_messages[0]
+    last_error_notification = ReportScheduleDAO.find_last_error_notification(
+        persisted_schedule
+    )
+    if error_notification_fails:
+        assert any("SMTP unavailable" in message for message in error_messages)
+        assert last_error_notification is None
+    else:
+        assert last_error_notification is not None
+    assert slack_should_use_v2_api_mock.call_count == 1
+    assert get_channels_with_search_mock.call_count == 1
+    email_mock.assert_called_once()
+
+
 @pytest.mark.usefixtures("create_report_slack_chart")
 def test_report_schedule_not_found(create_report_slack_chart):
     """
@@ -1974,20 +2180,21 @@ def test_report_schedule_success_grace(create_alert_slack_chart_success):
 
 
 @pytest.mark.usefixtures("create_alert_slack_chart_grace")
-@patch("superset.utils.slack.WebClient.files_upload")
+@patch("superset.reports.notifications.slack_channel_resolver.get_channels_with_search")
+@patch("superset.reports.notifications.slack.should_use_v2_api", return_value=True)
+@patch("superset.reports.notifications.slackv2.get_slack_client")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
-@patch("superset.reports.notifications.slack.get_slack_client")
 def test_report_schedule_success_grace_end(
-    slack_client_mock_class,
     screenshot_mock,
-    file_upload_mock,
+    slack_client_mock,
+    slack_should_use_v2_api_mock,
+    get_channels_with_search_mock,
     create_alert_slack_chart_grace,
 ):
-    """
-    ExecuteReport Command: Test report schedule on grace to noop
-    """
+    """A Slack alert leaving grace upgrades to v2 and sends successfully."""
 
     screenshot_mock.return_value = SCREENSHOT_FILE
+    slack_client = _configure_v2_upload_client(slack_client_mock.return_value)
 
     # set current time to after the grace period
     current_time = create_alert_slack_chart_grace.last_eval_dttm + timedelta(
@@ -2001,9 +2208,17 @@ def test_report_schedule_success_grace_end(
     channel_name = notification_targets[0]
     channel_id = "channel_id"
 
-    slack_client_mock_class.return_value.conversations_list.return_value = {
-        "channels": [{"id": channel_id, "name": channel_name}]
-    }
+    get_channels_with_search_mock.return_value = (
+        [
+            {
+                "id": channel_id,
+                "name": channel_name,
+                "is_member": True,
+                "is_private": True,
+            }
+        ],
+        False,
+    )
 
     with freeze_time(current_time):
         AsyncExecuteReportScheduleCommand(
@@ -2012,6 +2227,13 @@ def test_report_schedule_success_grace_end(
 
     db.session.commit()
     assert create_alert_slack_chart_grace.last_state == ReportState.SUCCESS
+    recipient = create_alert_slack_chart_grace.recipients[0]
+    assert recipient.type == ReportRecipientType.SLACKV2
+    assert json.loads(recipient.recipient_config_json) == {"target": channel_id}
+    slack_should_use_v2_api_mock.assert_called_once_with(raise_on_error=True)
+    completion_call = slack_client.files_completeUploadExternal.call_args
+    assert completion_call.kwargs["channel_id"] == channel_id
+    assert slack_client._upload_file.call_args.kwargs["data"] == SCREENSHOT_FILE
 
 
 @pytest.mark.usefixtures("create_alert_email_chart")
@@ -2165,7 +2387,7 @@ def test_slack_chart_alert_no_attachment(email_mock, create_alert_email_chart):
     "load_birth_names_dashboard_with_slices",
     "create_report_slack_chart",
 )
-@patch("superset.commands.report.execute.get_channels_with_search")
+@patch("superset.reports.notifications.slack_channel_resolver.get_channels_with_search")
 @patch("superset.utils.slack.WebClient")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
 def test_slack_token_callable_chart_report(
@@ -2182,20 +2404,24 @@ def test_slack_token_callable_chart_report(
     channel_name = notification_targets[0]
     channel_id = "channel_id"
     slack_client_mock_class.return_value = Mock()
+    _configure_v2_upload_client(slack_client_mock_class.return_value)
     # should_use_v2_api() probes via conversations_list(); a non-erroring return
     # is enough — it doesn't read the response body. The v2 upgrade then resolves
     # channel names through get_channels_with_search, which we mock directly.
     slack_client_mock_class.return_value.conversations_list.return_value = {
         "channels": [{"id": channel_id, "name": channel_name}]
     }
-    get_channels_with_search_mock.return_value = [
-        {
-            "id": channel_id,
-            "name": channel_name,
-            "is_member": True,
-            "is_private": False,
-        }
-    ]
+    get_channels_with_search_mock.return_value = (
+        [
+            {
+                "id": channel_id,
+                "name": channel_name,
+                "is_member": True,
+                "is_private": False,
+            }
+        ],
+        False,
+    )
 
     slack_token_mock = Mock(return_value="cool_code")
     with patch.dict("flask.current_app.config", {"SLACK_API_TOKEN": slack_token_mock}):
@@ -2207,11 +2433,20 @@ def test_slack_token_callable_chart_report(
                 TEST_ID, create_report_slack_chart.id, datetime.utcnow()
             ).run()
             slack_token_mock.assert_called()
-            slack_client_mock_class.assert_called_with(
-                token="cool_code",  # noqa: S106
-                proxy=None,
-                timeout=30,
-            )
+            assert slack_client_mock_class.call_args_list == [
+                call(
+                    token="cool_code",  # noqa: S106
+                    proxy=None,
+                    timeout=30,
+                    retry_handlers=None,
+                ),
+                call(
+                    token="cool_code",  # noqa: S106
+                    proxy=None,
+                    timeout=30,
+                    retry_handlers=[],
+                ),
+            ]
             assert_log(ReportState.SUCCESS)
 
 
@@ -2616,9 +2851,10 @@ def test_prune_log_soft_time_out(bulk_delete_logs, create_report_email_dashboard
 def test__send_with_client_errors(notification_mock, logger_mock):
     notification_content = "I am some content"
     recipients = ["test@foo.com"]
+    report_state = BaseReportState(ReportSchedule(), datetime.utcnow(), uuid4())
     notification_mock.return_value.send.side_effect = NotificationParamException()
     with pytest.raises(ReportScheduleClientErrorsException) as excinfo:
-        BaseReportState._send(BaseReportState, notification_content, recipients)
+        report_state._send(notification_content, recipients)
 
     assert excinfo.errisinstance(SupersetException)
     logger_mock.warning.assert_called_with(
@@ -2631,13 +2867,14 @@ def test__send_with_client_errors(notification_mock, logger_mock):
 def test__send_with_multiple_errors(notification_mock, logger_mock):
     notification_content = "I am some content"
     recipients = ["test@foo.com", "test2@bar.com"]
+    report_state = BaseReportState(ReportSchedule(), datetime.utcnow(), uuid4())
     notification_mock.return_value.send.side_effect = [
         NotificationParamException(),
         NotificationError(),
     ]
     # it raises the error with a 500 status if present
     with pytest.raises(ReportScheduleSystemErrorsException) as excinfo:
-        BaseReportState._send(BaseReportState, notification_content, recipients)
+        report_state._send(notification_content, recipients)
 
     assert excinfo.errisinstance(SupersetException)
     # it logs both errors as warnings
@@ -2658,9 +2895,10 @@ def test__send_with_multiple_errors(notification_mock, logger_mock):
 def test__send_with_server_errors(notification_mock, logger_mock):
     notification_content = "I am some content"
     recipients = ["test@foo.com"]
+    report_state = BaseReportState(ReportSchedule(), datetime.utcnow(), uuid4())
     notification_mock.return_value.send.side_effect = NotificationError()
     with pytest.raises(ReportScheduleSystemErrorsException) as excinfo:
-        BaseReportState._send(BaseReportState, notification_content, recipients)
+        report_state._send(notification_content, recipients)
 
     assert excinfo.errisinstance(SupersetException)
     # it logs the error

tests/unit_tests/utils/slack_test.py

@@ -16,33 +16,85 @@
 # under the License.
 
 import warnings
+from email.message import Message
+from http.client import RemoteDisconnected
+from urllib.error import HTTPError, URLError
 
 import pytest
 from slack_sdk.errors import (
     SlackApiError,
+    SlackClientConfigurationError,
     SlackClientNotConnectedError,
     SlackRequestError,
+    SlackTokenRotationError,
+)
+from slack_sdk.http_retry.builtin_handlers import (
+    ConnectionErrorRetryHandler,
+    RateLimitErrorRetryHandler,
 )
 
+from superset.constants import CACHE_DISABLED_TIMEOUT
+from superset.exceptions import SupersetException
 from superset.utils.slack import (
     _emit_v1_flag_off_deprecation,
     _emit_v1_scope_missing_deprecation,
     _SLACK_V1_DEPRECATION_MESSAGE,
+    get_channels,
     get_channels_with_search,
+    get_slack_client,
+    is_transient_slack_api_error,
+    refresh_cached_slack_channels_with_search,
     should_use_v2_api,
+    SlackChannelListingClientError,
+    SlackChannelListingError,
     SlackChannelTypes,
+    SlackV2ProbeClientError,
+    SlackV2ProbeError,
 )
 
 
 class MockResponse:
-    def __init__(self, data):
+    def __init__(self, data, status_code: int | None = None):
         self._data = data
+        self.status_code = status_code
 
     @property
     def data(self):
         return self._data
 
 
+def test_delivery_client_disables_outcome_unknown_connection_retries(mocker) -> None:
+    mocker.patch.dict(
+        "superset.utils.slack.app.config",
+        {
+            "SLACK_API_TOKEN": "xoxb-test",
+            "SLACK_PROXY": None,
+            "SLACK_API_TIMEOUT": 30,
+            "SLACK_API_RATE_LIMIT_RETRY_COUNT": 2,
+        },
+    )
+
+    delivery_client = get_slack_client(for_delivery=True)
+    discovery_client = get_slack_client()
+
+    assert not any(
+        isinstance(handler, RateLimitErrorRetryHandler)
+        for handler in delivery_client.retry_handlers
+    )
+    assert not any(
+        isinstance(handler, ConnectionErrorRetryHandler)
+        for handler in delivery_client.retry_handlers
+    )
+    assert any(
+        isinstance(handler, ConnectionErrorRetryHandler)
+        for handler in discovery_client.retry_handlers
+    )
+    assert any(
+        isinstance(handler, RateLimitErrorRetryHandler)
+        for handler in discovery_client.retry_handlers
+    )
+
+
 class TestGetChannelsWithSearch:
     # Fetch all channels when no search string is provided
     def test_fetch_all_channels_no_search_string(self, mocker):
@@ -100,6 +152,20 @@ class TestGetChannelsWithSearch:
         # Assert that the result is a list with a single channel dictionary
         assert result == [{"name": "general", "id": "C12345"}]
 
+    def test_exact_match_uses_unicode_casefolding(self, mocker) -> None:
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = MockResponse(
+            {
+                "channels": [{"name": "Straße", "id": "C12345"}],
+                "response_metadata": {"next_cursor": None},
+            }
+        )
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(search_string="STRASSE", exact_match=True)
+
+        assert result == [{"name": "Straße", "id": "C12345"}]
+
     def test_handle_exact_match_search_string_multiple_channels(self, mocker):
         mock_data = {
             "channels": [
@@ -164,6 +230,29 @@ class TestGetChannelsWithSearch:
 The server responded with: missing scope: channels:read"""
         )
 
+    @pytest.mark.parametrize(
+        ("error_code", "expected_exception"),
+        [
+            ("invalid_auth", SlackChannelListingClientError),
+            ("service_unavailable", SlackChannelListingError),
+        ],
+    )
+    def test_channel_listing_preserves_slack_error_classification(
+        self,
+        mocker,
+        error_code: str,
+        expected_exception: type[Exception],
+    ) -> None:
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = SlackApiError(
+            "listing failed",
+            {"ok": False, "error": error_code},
+        )
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with pytest.raises(expected_exception):
+            get_channels_with_search(force=True)
+
     @pytest.mark.parametrize(
         "types, expected_channel_ids",
         [
@@ -224,6 +313,486 @@ The server responded with: missing scope: channels:read"""
 
         assert mock_client.conversations_list.call_args.kwargs["team_id"] == "T123456"
 
+    def test_cache_hit_and_channels_come_from_one_read(self, mocker) -> None:
+        cached_channels = [{"id": "C1", "name": "cached"}]
+        cache_get = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=cached_channels,
+        )
+        channel_fetch = mocker.patch("superset.utils.slack._get_channels")
+        mocker.patch.dict(
+            "superset.utils.slack.app.config", {"SLACK_TEAM_ID": "T123456"}
+        )
+
+        channels, used_cache = get_channels_with_search(return_cache_status=True)
+
+        assert channels == cached_channels
+        assert used_cache is True
+        cache_get.assert_called_once_with("slack_conversations_list_T123456")
+        channel_fetch.assert_not_called()
+
+    def test_disabled_cache_default_search_ignores_stale_entry(self, mocker) -> None:
+        cache_get = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=[{"id": "C0", "name": "stale"}],
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            side_effect=[
+                [{"id": "C1", "name": "first"}],
+                [{"id": "C2", "name": "second"}],
+            ],
+        )
+        mocker.patch.dict(
+            "superset.utils.slack.app.config",
+            {"SLACK_CACHE_TIMEOUT": CACHE_DISABLED_TIMEOUT},
+        )
+
+        assert get_channels_with_search() == [{"id": "C1", "name": "first"}]
+        assert get_channels_with_search() == [{"id": "C2", "name": "second"}]
+
+        cache_get.assert_not_called()
+        cache_set.assert_not_called()
+        assert channel_fetch.call_count == 2
+
+    def test_cache_miss_fetches_once_and_reports_live_provenance(self, mocker) -> None:
+        live_channels = [{"id": "C2", "name": "live"}]
+        cache_get = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=None,
+        )
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+
+        channels, used_cache = get_channels_with_search(return_cache_status=True)
+
+        assert channels == live_channels
+        assert used_cache is False
+        cache_get.assert_called_once_with("slack_conversations_list")
+        channel_fetch.assert_called_once_with(
+            "slack_conversations_list",
+            team_id=None,
+            cache=False,
+        )
+        cache_set.assert_called_once_with(
+            "slack_conversations_list",
+            live_channels,
+            timeout=mocker.ANY,
+        )
+
+    def test_cache_read_failure_falls_back_to_live_channels(self, mocker) -> None:
+        live_channels = [{"id": "C2", "name": "live"}]
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            side_effect=ConnectionError("Redis unavailable"),
+        )
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        logger = mocker.patch("superset.utils.slack.logger")
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+
+        channels, used_cache = get_channels_with_search(return_cache_status=True)
+
+        assert channels == live_channels
+        assert used_cache is False
+        channel_fetch.assert_called_once()
+        cache_set.assert_called_once()
+        logger.warning.assert_called_once()
+
+    def test_default_search_cache_read_failure_uses_live_channels(self, mocker) -> None:
+        live_channels = [{"id": "C2", "name": "live"}]
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            side_effect=ConnectionError("Redis unavailable"),
+        )
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        logger = mocker.patch("superset.utils.slack.logger")
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+
+        assert get_channels_with_search() == live_channels
+
+        channel_fetch.assert_called_once()
+        cache_set.assert_called_once()
+        logger.warning.assert_called_once()
+
+    def test_forced_search_cache_write_failure_preserves_live_channels(
+        self, mocker
+    ) -> None:
+        live_channels = [{"id": "C2", "name": "live"}]
+        cache_get = mocker.patch("superset.utils.slack.cache_manager.cache.get")
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.set",
+            side_effect=ConnectionError("Redis unavailable"),
+        )
+        logger = mocker.patch("superset.utils.slack.logger")
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+
+        assert get_channels_with_search(force=True) == live_channels
+
+        cache_get.assert_not_called()
+        channel_fetch.assert_called_once()
+        logger.warning.assert_called_once()
+
+    def test_cache_write_failure_preserves_live_channels(self, mocker) -> None:
+        live_channels = [{"id": "C2", "name": "live"}]
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=None,
+        )
+        mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.set",
+            side_effect=ConnectionError("Redis unavailable"),
+        )
+        logger = mocker.patch("superset.utils.slack.logger")
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+
+        channels, used_cache = get_channels_with_search(return_cache_status=True)
+
+        assert channels == live_channels
+        assert used_cache is False
+        logger.warning.assert_called_once()
+
+    def test_metastore_cache_miss_fetches_without_cache_write(self, mocker) -> None:
+        live_channels = [{"id": "C2", "name": "live"}]
+        mocker.patch("superset.utils.slack.cache_manager.cache.get", return_value=None)
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=True,
+        )
+
+        channels, used_cache = get_channels_with_search(return_cache_status=True)
+
+        assert channels == live_channels
+        assert used_cache is False
+        channel_fetch.assert_called_once_with(
+            "slack_conversations_list",
+            team_id=None,
+            cache=False,
+        )
+        cache_set.assert_not_called()
+
+    def test_forced_warmup_writes_metastore_channel_cache(self, mocker) -> None:
+        live_channels = [{"id": "C2", "name": "live"}]
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=True,
+        )
+
+        assert get_channels(force=True, cache_timeout=123) == live_channels
+
+        channel_fetch.assert_called_once_with(
+            "slack_conversations_list",
+            team_id=None,
+            cache=False,
+        )
+        cache_set.assert_called_once_with(
+            "slack_conversations_list",
+            live_channels,
+            timeout=123,
+        )
+
+    @pytest.mark.parametrize("cache_set_result", [True, None])
+    def test_refreshes_cached_channels_once_per_workspace_cooldown(
+        self, mocker, cache_set_result
+    ) -> None:
+        refreshed_channels = [{"id": "C2", "name": "new", "is_private": False}]
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+        channel_search = mocker.patch(
+            "superset.utils.slack.get_channels_with_search",
+            return_value=refreshed_channels,
+        )
+        cache_get = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=None,
+        )
+        cache_set = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.set",
+            return_value=cache_set_result,
+        )
+        mocker.patch.dict(
+            "superset.utils.slack.app.config", {"SLACK_TEAM_ID": "T123456"}
+        )
+
+        assert (
+            refresh_cached_slack_channels_with_search(
+                search_string="new",
+                types=[SlackChannelTypes.PUBLIC],
+                exact_match=True,
+            )
+            == refreshed_channels
+        )
+
+        channel_search.assert_called_once_with(
+            force=True,
+            cache=False,
+        )
+        cache_get.assert_called_once_with(
+            "slack_conversations_list_T123456_refresh_cooldown"
+        )
+        assert cache_set.call_args_list == [
+            mocker.call(
+                "slack_conversations_list_T123456",
+                refreshed_channels,
+                timeout=mocker.ANY,
+            ),
+            mocker.call(
+                "slack_conversations_list_T123456_refresh_cooldown",
+                True,
+                timeout=300,
+            ),
+        ]
+
+    def test_failed_channel_cache_write_does_not_record_cooldown(self, mocker) -> None:
+        refreshed_channels = [{"id": "C2", "name": "new"}]
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+        channel_search = mocker.patch(
+            "superset.utils.slack.get_channels_with_search",
+            return_value=refreshed_channels,
+        )
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=None,
+        )
+        cache_set = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.set",
+            return_value=False,
+        )
+
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new"}
+        ]
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new"}
+        ]
+
+        assert channel_search.call_count == 2
+        assert cache_set.call_args_list == [
+            mocker.call(
+                "slack_conversations_list",
+                refreshed_channels,
+                timeout=mocker.ANY,
+            ),
+            mocker.call(
+                "slack_conversations_list",
+                refreshed_channels,
+                timeout=mocker.ANY,
+            ),
+        ]
+
+    def test_disabled_cache_refresh_ignores_stale_cooldown(self, mocker) -> None:
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+        cache_get = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=True,
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        channel_search = mocker.patch(
+            "superset.utils.slack.get_channels_with_search",
+            return_value=[{"id": "C2", "name": "new"}],
+        )
+        mocker.patch.dict(
+            "superset.utils.slack.app.config",
+            {"SLACK_CACHE_TIMEOUT": CACHE_DISABLED_TIMEOUT},
+        )
+
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new"}
+        ]
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new"}
+        ]
+
+        assert channel_search.call_args_list == [
+            mocker.call(
+                search_string="new",
+                types=None,
+                exact_match=False,
+                force=True,
+                cache=False,
+            ),
+            mocker.call(
+                search_string="new",
+                types=None,
+                exact_match=False,
+                force=True,
+                cache=False,
+            ),
+        ]
+        cache_get.assert_not_called()
+        cache_set.assert_not_called()
+
+    def test_recent_refresh_uses_cached_channels_without_another_force(
+        self, mocker
+    ) -> None:
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+        mocker.patch("superset.utils.slack.cache_manager.cache.get", return_value=True)
+        cached_channels = mocker.patch(
+            "superset.utils.slack._get_channels_with_cache_status",
+            return_value=(
+                [{"id": "C2", "name": "new", "is_private": False}],
+                True,
+            ),
+        )
+
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new", "is_private": False}
+        ]
+
+        cached_channels.assert_called_once_with()
+
+    def test_refresh_cooldown_read_failure_uses_live_channels(self, mocker) -> None:
+        refreshed_channels = [{"id": "C2", "name": "new", "is_private": False}]
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            side_effect=ConnectionError("Redis unavailable"),
+        )
+        channel_search = mocker.patch(
+            "superset.utils.slack.get_channels_with_search",
+            return_value=refreshed_channels,
+        )
+        logger = mocker.patch("superset.utils.slack.logger")
+
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new", "is_private": False}
+        ]
+
+        channel_search.assert_called_once_with(force=True, cache=False)
+        logger.warning.assert_called_once()
+
+    def test_recent_refresh_survives_channel_cache_read_failure(self, mocker) -> None:
+        live_channels = [{"id": "C2", "name": "new", "is_private": False}]
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+        cache_get = mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            side_effect=[True, ConnectionError("Redis unavailable")],
+        )
+        channel_fetch = mocker.patch(
+            "superset.utils.slack._get_channels",
+            return_value=live_channels,
+        )
+        mocker.patch("superset.utils.slack.cache_manager.cache.set")
+
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new", "is_private": False}
+        ]
+
+        assert cache_get.call_count == 2
+        channel_fetch.assert_called_once()
+
+    def test_session_backed_cache_uses_uncached_refresh(self, mocker) -> None:
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=True,
+        )
+        cache_get = mocker.patch("superset.utils.slack.cache_manager.cache.get")
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+        channel_search = mocker.patch(
+            "superset.utils.slack.get_channels_with_search",
+            return_value=[{"id": "C2", "name": "new"}],
+        )
+
+        assert refresh_cached_slack_channels_with_search(search_string="new") == [
+            {"id": "C2", "name": "new"}
+        ]
+
+        channel_search.assert_called_once_with(
+            search_string="new",
+            types=None,
+            exact_match=False,
+            force=True,
+            cache=False,
+        )
+        cache_get.assert_not_called()
+        cache_set.assert_not_called()
+
+    def test_failed_forced_refresh_does_not_record_cooldown(self, mocker) -> None:
+        mocker.patch(
+            "superset.utils.slack._slack_channel_cache_uses_report_session",
+            return_value=False,
+        )
+        channel_search = mocker.patch(
+            "superset.utils.slack.get_channels_with_search",
+            side_effect=SupersetException("Slack listing failed"),
+        )
+        mocker.patch(
+            "superset.utils.slack.cache_manager.cache.get",
+            return_value=None,
+        )
+        cache_set = mocker.patch("superset.utils.slack.cache_manager.cache.set")
+
+        with pytest.raises(SupersetException, match="Slack listing failed"):
+            refresh_cached_slack_channels_with_search(search_string="new")
+
+        channel_search.assert_called_once_with(
+            force=True,
+            cache=False,
+        )
+        cache_set.assert_not_called()
+
     def test_resolves_callable_team_id(self, mocker):
         # SLACK_TEAM_ID may be a callable (e.g. to fetch from a secrets store),
         # mirroring SLACK_API_TOKEN; it is resolved before being forwarded.
@@ -309,9 +878,9 @@ The server responded with: missing scope: channels:read"""
 def _reset_v1_warning_caches():
     """Each test sees fresh once-per-process warning state.
 
-    The deprecation emitters are wrapped in `functools.cache` to give
-    thread-safe one-shot semantics in production. Tests need them to fire
-    again, so we clear the cache before and after each case.
+    The deprecation emitters use `functools.cache` to suppress calls after the
+    first one completes. Tests need them to fire again, so we clear the cache
+    before and after each case.
     """
     _emit_v1_flag_off_deprecation.cache_clear()
     _emit_v1_scope_missing_deprecation.cache_clear()
@@ -384,7 +953,7 @@ class TestShouldUseV2Api:
 
         with warnings.catch_warnings(record=True) as caught:
             warnings.simplefilter("always")
-            assert should_use_v2_api() is False
+            assert should_use_v2_api(raise_on_error=True) is False
             assert should_use_v2_api() is False
             assert should_use_v2_api() is False
 
@@ -433,6 +1002,19 @@ class TestShouldUseV2Api:
         assert logger_mock.warning.call_count == 1
         assert "channels:read" in logger_mock.warning.call_args.args[0]
 
+    @pytest.mark.parametrize("status_code", [429, 503])
+    def test_transient_probe_error_detected_from_http_status(
+        self,
+        status_code: int,
+    ) -> None:
+        """Rate-limit and server HTTP statuses remain system failures."""
+        error = SlackApiError(
+            message="probe failed",
+            response=MockResponse({"ok": False}, status_code=status_code),
+        )
+
+        assert is_transient_slack_api_error(error, "") is True
+
     @pytest.mark.parametrize(
         "error_code",
         ["invalid_auth", "ratelimited", "fatal_error", "account_inactive", ""],
@@ -478,9 +1060,10 @@ class TestShouldUseV2Api:
         [
             SlackClientNotConnectedError("transport closed"),
             SlackRequestError("bad request args"),
+            URLError("connection reset"),
         ],
     )
-    def test_returns_false_on_slack_sdk_client_error_from_probe(
+    def test_returns_false_on_slack_client_or_transport_error_from_probe(
         self, exception: Exception, mocker
     ):
         """Non-`SlackApiError` SDK failures (e.g. `SlackClientNotConnectedError`,
@@ -502,7 +1085,78 @@ class TestShouldUseV2Api:
 
         assert should_use_v2_api() is False
         assert logger_mock.warning.call_count == 1
-        assert "probe failed to connect" in logger_mock.warning.call_args.args[0]
+        assert "probe failed" in logger_mock.warning.call_args.args[0]
+
+    @pytest.mark.parametrize(
+        "exception",
+        [
+            SlackApiError(
+                message="service unavailable",
+                response={"ok": False, "error": "service_unavailable"},
+            ),
+            SlackApiError(
+                message="rate limited",
+                response={"ok": False, "error": "ratelimited"},
+            ),
+            SlackClientNotConnectedError("transport closed"),
+            URLError("connection reset"),
+            ConnectionResetError("connection reset"),
+            RemoteDisconnected("connection closed"),
+            TimeoutError("timed out"),
+            HTTPError("https://slack.com", 504, "unavailable", Message(), None),
+        ],
+    )
+    def test_raises_system_error_for_transient_probe_when_requested(
+        self,
+        exception: Exception,
+        mocker,
+    ) -> None:
+        """File sends preserve transient probe failures for system handling."""
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=True,
+        )
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = exception
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with pytest.raises(SlackV2ProbeError, match="probe failed"):
+            should_use_v2_api(raise_on_error=True)
+
+    @pytest.mark.parametrize(
+        "exception",
+        [
+            SlackApiError(
+                message="invalid auth",
+                response={"ok": False, "error": "invalid_auth"},
+            ),
+            SlackRequestError("bad request args"),
+            SlackClientConfigurationError("invalid client configuration"),
+            SlackTokenRotationError("token rotation failed"),
+            HTTPError("https://slack.com", 413, "too large", Message(), None),
+        ],
+    )
+    def test_raises_client_error_for_permanent_probe_failure_when_requested(
+        self,
+        exception: Exception,
+        mocker,
+    ) -> None:
+        """Permanent probe failures remain operator-fixable client errors."""
+        mocker.patch(
+            "superset.utils.slack.feature_flag_manager.is_feature_enabled",
+            return_value=True,
+        )
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = exception
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        with pytest.raises(SlackV2ProbeClientError, match="probe failed") as exc_info:
+            should_use_v2_api(raise_on_error=True)
+
+        assert exc_info.value.status == 422
+
+    def test_client_probe_error_has_common_probe_error_parent(self) -> None:
+        assert issubclass(SlackV2ProbeClientError, SlackV2ProbeError)
 
     def test_propagates_non_sdk_errors_from_probe(self, mocker):
         """A truly unexpected, non-SDK exception (e.g. a programming error) still

tests/unit_tests/utils/test_decorators.py

@@ -36,6 +36,10 @@ class ResponseValues(StrEnum):
     OK = "ok"
 
 
+class WarningError(Exception):
+    status = 400
+
+
 def test_debounce() -> None:
     mock = Mock()
 
@@ -63,7 +67,7 @@ def test_debounce() -> None:
     [
         (ResponseValues.OK, None, "custom.prefix.ok"),
         (ResponseValues.FAIL, ValueError, "custom.prefix.error"),
-        (ResponseValues.WARN, FileNotFoundError, "custom.prefix.warn"),
+        (ResponseValues.WARN, WarningError, "custom.prefix.warning"),
     ],
 )
 def test_statsd_gauge(
@@ -74,7 +78,7 @@ def test_statsd_gauge(
         if response == ResponseValues.FAIL:
             raise ValueError("Error")
         if response == ResponseValues.WARN:
-            raise FileNotFoundError("Not found")
+            raise WarningError("Warning")
         return "OK"
 
     with patch("superset.extensions.stats_logger_manager.instance.gauge") as mock:
@@ -86,7 +90,45 @@ def test_statsd_gauge(
 
         with cm:
             my_func(response_value, 1, 2)
-            mock.assert_called_once_with(expected_result, 1)
+
+        mock.assert_called_once_with(expected_result, 1)
+
+
+def test_statsd_gauge_ignores_configured_exception() -> None:
+    class RoutingSignalError(Exception):
+        pass
+
+    @decorators.statsd_gauge(
+        "custom.prefix",
+        ignored_exceptions=(RoutingSignalError,),
+    )
+    def my_func() -> None:
+        raise RoutingSignalError
+
+    with (
+        patch("superset.extensions.stats_logger_manager.instance.gauge") as mock,
+        pytest.raises(RoutingSignalError),
+    ):
+        my_func()
+
+    mock.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("exception", "expected_metric"),
+    [
+        (ValueError("failure"), "custom.prefix.error"),
+        (WarningError("warning"), "custom.prefix.warning"),
+    ],
+)
+def test_record_statsd_gauge_failure_uses_shared_severity_contract(
+    exception: Exception,
+    expected_metric: str,
+) -> None:
+    with patch("superset.extensions.stats_logger_manager.instance.gauge") as mock:
+        decorators.record_statsd_gauge_failure("custom.prefix", exception)
+
+    mock.assert_called_once_with(expected_metric, 1)
 
 
 @patch("superset.utils.decorators.g")