From 83f86c26644a3738f672282c65005976b41ad78d Mon Sep 17 00:00:00 2001 From: Mateusz Kuprowski Date: Tue, 14 Jul 2026 16:43:35 +0200 Subject: [PATCH 1/2] fix(FS-2139): centralize Slack rate-limit retries --- CHANGELOG.md | 6 + test/unit/connectors/test_slack.py | 155 +++++++++++++++++- unstructured_ingest/__version__.py | 2 +- .../processes/connectors/slack.py | 34 +++- 4 files changed, 193 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 95b316dea..0ca1785a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## [1.6.29] + +### Fixes + +- **fix(FS-2139): centralize bounded Slack API rate-limit retries in SDK client configuration.** Sync and async Slack clients now use the Slack SDK's connection and rate-limit retry handlers configured once in `SlackConnectionConfig`, replacing custom call-site retry loops across indexer join/history and downloader history/replies/files calls. + ## [1.6.28] ### Fixes diff --git a/test/unit/connectors/test_slack.py b/test/unit/connectors/test_slack.py index fb7c031fd..903e6bdb6 100644 --- a/test/unit/connectors/test_slack.py +++ b/test/unit/connectors/test_slack.py @@ -1,8 +1,11 @@ import hashlib from datetime import datetime, timezone -from unittest.mock import Mock +from io import BytesIO +from unittest.mock import Mock, call +from urllib.error import HTTPError import pytest +from pydantic import Secret from unstructured_ingest.data_types.file_data import ( FileData, @@ -13,13 +16,17 @@ from unstructured_ingest.error import ValueError as IngestValueError from unstructured_ingest.processes.connectors.slack import ( PRIVATE_FILE_DOWNLOAD_TIMEOUT_SECONDS, + SLACK_RATE_LIMIT_MAX_RETRIES, SlackAccessConfig, + SlackConnectionConfig, SlackDownloader, SlackIndexer, SlackIndexerConfig, _channel_history_error_msg, _channel_join_error_msg, _NoRedirectHandler, + _slack_async_retry_handlers, + _slack_sync_retry_handlers, _token_kind, ) @@ -347,6 +354,152 @@ def _make_slack_api_error(error_code: str): return SlackApiError(message=error_code, response=response) +def _make_rate_limit_http_error() -> HTTPError: + from email.message import Message + + headers = Message() + headers.add_header("Retry-After", "3") + return HTTPError( + url="https://slack.com/api/conversations.join", + code=429, + msg="Too Many Requests", + hdrs=headers, + fp=BytesIO(b'{"ok":false,"error":"rate_limited"}'), + ) + + +def _raise_rate_limit_http_error(*_args, **_kwargs) -> HTTPError: + raise _make_rate_limit_http_error() + + +def test_slack_sync_retry_handlers_include_connection_and_rate_limit_handlers(): + from slack_sdk.http_retry import ConnectionErrorRetryHandler, RateLimitErrorRetryHandler + + handlers = _slack_sync_retry_handlers() + + assert len(handlers) == 2 + assert isinstance(handlers[0], ConnectionErrorRetryHandler) + assert isinstance(handlers[1], RateLimitErrorRetryHandler) + assert handlers[1].max_retry_count == SLACK_RATE_LIMIT_MAX_RETRIES + + +def test_slack_async_retry_handlers_include_connection_and_rate_limit_handlers(): + from slack_sdk.http_retry.builtin_async_handlers import ( + AsyncConnectionErrorRetryHandler, + AsyncRateLimitErrorRetryHandler, + ) + + handlers = _slack_async_retry_handlers() + + assert len(handlers) == 2 + assert isinstance(handlers[0], AsyncConnectionErrorRetryHandler) + assert isinstance(handlers[1], AsyncRateLimitErrorRetryHandler) + assert handlers[1].max_retry_count == SLACK_RATE_LIMIT_MAX_RETRIES + + +def test_slack_connection_config_get_client_configures_retry_handlers(): + from slack_sdk.http_retry import ConnectionErrorRetryHandler, RateLimitErrorRetryHandler + + config = SlackConnectionConfig(access_config=Secret(SlackAccessConfig(token=BOT_TOKEN))) + client = config.get_client() + + assert len(client.retry_handlers) == 2 + assert isinstance(client.retry_handlers[0], ConnectionErrorRetryHandler) + assert isinstance(client.retry_handlers[1], RateLimitErrorRetryHandler) + assert client.retry_handlers[1].max_retry_count == SLACK_RATE_LIMIT_MAX_RETRIES + + +def test_slack_connection_config_get_async_client_configures_retry_handlers(): + from slack_sdk.http_retry.builtin_async_handlers import ( + AsyncConnectionErrorRetryHandler, + AsyncRateLimitErrorRetryHandler, + ) + + config = SlackConnectionConfig(access_config=Secret(SlackAccessConfig(token=BOT_TOKEN))) + client = config.get_async_client() + + assert len(client.retry_handlers) == 2 + assert isinstance(client.retry_handlers[0], AsyncConnectionErrorRetryHandler) + assert isinstance(client.retry_handlers[1], AsyncRateLimitErrorRetryHandler) + assert client.retry_handlers[1].max_retry_count == SLACK_RATE_LIMIT_MAX_RETRIES + + +def test_slack_sync_client_honors_retry_after_with_jitter_until_retries_exhausted(mocker): + from slack_sdk.errors import SlackApiError + + sleep = mocker.patch("slack_sdk.http_retry.builtin_handlers.time.sleep") + mocker.patch("slack_sdk.http_retry.builtin_handlers.random.random", return_value=0.25) + mocker.patch( + "slack_sdk.web.base_client.BaseClient._perform_urllib_http_request_internal", + side_effect=_raise_rate_limit_http_error, + ) + + config = SlackConnectionConfig(access_config=Secret(SlackAccessConfig(token=BOT_TOKEN))) + client = config.get_client() + + with pytest.raises(SlackApiError): + client.conversations_join(channel="C1") + + assert sleep.call_args_list == [call(3.25)] * SLACK_RATE_LIMIT_MAX_RETRIES + + +@pytest.mark.asyncio +async def test_slack_async_client_honors_retry_after_with_jitter_until_retries_exhausted(mocker): + from slack_sdk.errors import SlackApiError + + sleep = mocker.patch("slack_sdk.http_retry.builtin_async_handlers.asyncio.sleep") + mocker.patch("slack_sdk.http_retry.builtin_async_handlers.random.random", return_value=0.25) + + class MockResponse: + status = 429 + headers = {"Retry-After": "3"} + content_type = "application/json" + + async def json(self): + return {"ok": False, "error": "rate_limited"} + + async def text(self): + return '{"ok": false, "error": "rate_limited"}' + + async def read(self): + return b'{"ok": false, "error": "rate_limited"}' + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + session = mocker.Mock() + session.closed = False + session.request = mocker.Mock(return_value=MockResponse()) + session.close = mocker.AsyncMock() + + config = SlackConnectionConfig(access_config=Secret(SlackAccessConfig(token=BOT_TOKEN))) + client = config.get_async_client() + client.session = session + connection_config = mocker.Mock() + connection_config.get_async_client.return_value = client + downloader = SlackDownloader(connection_config=connection_config) + file_data = FileData( + identifier="pkg-1", + connector_type="slack", + source_identifiers=SourceIdentifiers(filename="abc.xml", fullpath="abc.xml"), + metadata=FileDataSourceMetadata( + record_locator={ + "channel": CHANNEL, + "oldest": "1.0", + "latest": "2.0", + } + ), + ) + + with pytest.raises(SlackApiError): + await downloader._download_conversation(file_data, Mock()) + + assert sleep.call_args_list == [call(3.25)] * SLACK_RATE_LIMIT_MAX_RETRIES + + def test_validate_channels_bot_succeeds_when_all_joins_succeed(): client = Mock() indexer = _make_indexer(channels=["C1", "C2"]) diff --git a/unstructured_ingest/__version__.py b/unstructured_ingest/__version__.py index 5a23218dd..d2e431858 100644 --- a/unstructured_ingest/__version__.py +++ b/unstructured_ingest/__version__.py @@ -1 +1 @@ -__version__ = "1.6.28" # pragma: no cover +__version__ = "1.6.29" # pragma: no cover diff --git a/unstructured_ingest/processes/connectors/slack.py b/unstructured_ingest/processes/connectors/slack.py index 33def2ad9..22e35abdf 100644 --- a/unstructured_ingest/processes/connectors/slack.py +++ b/unstructured_ingest/processes/connectors/slack.py @@ -45,6 +45,30 @@ SLACK_PRIVATE_FILE_HOST = "files.slack.com" CONNECTOR_TYPE = "slack" +# The SDK excludes the initial request from max_retry_count, so 9 means 10 total attempts. +# Its 429 handlers honor Slack's Retry-After header and add 0–1 seconds of jitter. +SLACK_RATE_LIMIT_MAX_RETRIES = 9 + + +def _slack_sync_retry_handlers(): + from slack_sdk.http_retry import ConnectionErrorRetryHandler, RateLimitErrorRetryHandler + + return [ + ConnectionErrorRetryHandler(), + RateLimitErrorRetryHandler(max_retry_count=SLACK_RATE_LIMIT_MAX_RETRIES), + ] + + +def _slack_async_retry_handlers(): + from slack_sdk.http_retry.builtin_async_handlers import ( + AsyncConnectionErrorRetryHandler, + AsyncRateLimitErrorRetryHandler, + ) + + return [ + AsyncConnectionErrorRetryHandler(), + AsyncRateLimitErrorRetryHandler(max_retry_count=SLACK_RATE_LIMIT_MAX_RETRIES), + ] def _token_kind(token: str) -> Literal["user", "bot"]: @@ -185,14 +209,20 @@ class SlackConnectionConfig(ConnectionConfig): def get_client(self) -> "WebClient": from slack_sdk import WebClient - return WebClient(token=self.access_config.get_secret_value().token) + return WebClient( + token=self.access_config.get_secret_value().token, + retry_handlers=_slack_sync_retry_handlers(), + ) @requires_dependencies(["slack_sdk"], extras="slack") @SourceConnectionError.wrap def get_async_client(self) -> "AsyncWebClient": from slack_sdk.web.async_client import AsyncWebClient - return AsyncWebClient(token=self.access_config.get_secret_value().token) + return AsyncWebClient( + token=self.access_config.get_secret_value().token, + retry_handlers=_slack_async_retry_handlers(), + ) class SlackIndexerConfig(IndexerConfig): From 70e1faee789f4f66ab2695275ade58f0c577a7dc Mon Sep 17 00:00:00 2001 From: Mateusz Kuprowski Date: Tue, 14 Jul 2026 17:38:06 +0200 Subject: [PATCH 2/2] fix(FS-2139): install Slack async test dependencies Co-authored-by: Cursor --- pyproject.toml | 2 +- uv.lock | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index dfa07b24e..2756d3505 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -151,7 +151,7 @@ test = [ "beautifulsoup4", "pandas", # Connector specific deps - "slack_sdk", + "slack_sdk[optional]", "cryptography", "fsspec", "vertexai", diff --git a/uv.lock b/uv.lock index e39c2acc8..9680d352d 100644 --- a/uv.lock +++ b/uv.lock @@ -7355,7 +7355,7 @@ test = [ { name = "pytest-tagging" }, { name = "pytest-timeout" }, { name = "pytest-xdist" }, - { name = "slack-sdk" }, + { name = "slack-sdk", extra = ["optional"] }, { name = "universal-pathlib" }, { name = "vertexai" }, ] @@ -7532,7 +7532,7 @@ test = [ { name = "pytest-tagging" }, { name = "pytest-timeout" }, { name = "pytest-xdist" }, - { name = "slack-sdk" }, + { name = "slack-sdk", extras = ["optional"] }, { name = "universal-pathlib" }, { name = "vertexai" }, ]