fix(data-imports): recognize DeltaError as a transient object-store error

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={})

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