diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py index b769cd8ee556..c8d629d4eca9 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py @@ -5,7 +5,7 @@ from posthog.test.base import BaseTest from unittest.mock import AsyncMock, MagicMock, patch -import deltalake +import deltalake.exceptions from asgiref.sync import async_to_sync from parameterized import parameterized @@ -172,24 +172,38 @@ async def test_swallows_maintenance_failure(self): [ ( "credentials_loading", + OSError, "Operation not supported: an error occurred while loading credentials: dispatch failure: timeout", ), ( "credential_provider_not_enabled", + OSError, "Operation not supported: the credential provider was not enabled: no providers in chain provided credentials", ), ( "generic_s3_error", + OSError, "Generic S3 error: Error getting list response body: operation timed out", ), + ( + # table.vacuum()/optimize.compact() surface the identical object-store error text + # wrapped in DeltaError instead of OSError (unlike is_deltatable()'s OSError) — the + # exact shape of the issue this test guards against. + "generic_s3_error_as_delta_error", + deltalake.exceptions.DeltaError, + "Generic error: Kernel error: Error interacting with object store: Generic S3 error: " + "Server returned non-2xx status code: 503 Service Unavailable: SlowDown", + ), ] ) @pytest.mark.asyncio - async def test_logs_transient_object_store_error_without_capturing(self, _name: str, error_message: str): + async def test_logs_transient_object_store_error_without_capturing( + self, _name: str, error_cls: type[Exception], error_message: str + ): # A transient blip talking to our own delta S3 bucket (credential-provider or connectivity # errors from delta-rs) isn't a bug in this function — it shouldn't flood error tracking the # way an actual maintenance bug does (see test_swallows_maintenance_failure above). - helper = MagicMock(run_maintenance=AsyncMock(side_effect=OSError(error_message))) + helper = MagicMock(run_maintenance=AsyncMock(side_effect=error_cls(error_message))) logger = MagicMock(aexception=AsyncMock(), awarning=AsyncMock()) schema = MagicMock(partition_count=5, sync_type_config={}) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py index ea946a7d16e4..5ac2fa91e912 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py @@ -51,8 +51,8 @@ DEFAULT_COMPACT_FILES_PER_PARTITION_THRESHOLD = 200 DEFAULT_COMPACT_TOTAL_FILES_THRESHOLD = 5000 -# Substrings of the `OSError`s raised talking to our own S3-backed data-warehouse bucket that are -# transient and self-recovering, not a bug in our code or a customer credential problem: +# Substrings of the object-store errors raised talking to our own S3-backed data-warehouse bucket +# that are transient and self-recovering, not a bug in our code or a customer credential problem: # - the first three come from delta-rs's Rust `object_store` crate inside `DeltaTable.is_deltatable()` # (IMDS/STS blips, dispatch timeouts) # - "Please reduce your request rate" is S3's SlowDown throttling response, surfaced by s3fs/aiobotocore @@ -68,7 +68,16 @@ def is_transient_object_store_error(error: BaseException) -> bool: - return isinstance(error, OSError) and any(needle in str(error) for needle in TRANSIENT_OBJECT_STORE_ERRORS) + """True for a transient object-store error, however delta-rs happened to surface it. + + `DeltaTable.is_deltatable()` raises these as a plain `OSError`, but table-level operations + (e.g. `vacuum()`, `optimize.compact()`) wrap the identical underlying object-store error text in + `deltalake.exceptions.DeltaError` instead — same blip, different exception type depending on + which delta-rs entry point hit it. + """ + return isinstance(error, OSError | deltalake.exceptions.DeltaError) and any( + needle in str(error) for needle in TRANSIENT_OBJECT_STORE_ERRORS + ) # Delta's conflict checker raises CommitFailedError the moment a concurrent commit invalidates what