fix(data-imports): keep known-transient S3 blips out of error tracking in get_delta_table

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py

@@ -279,6 +279,17 @@ def get_storage_options(self) -> dict[str, str]:
         """Public accessor for the delta-rs storage options (used by the in-place repartitioner)."""
         return self._get_credentials()
 
+    async def _capture_unless_transient(self, e: Exception) -> None:
+        """capture_exception unless `e` is a known-transient object-store blip (see
+        is_transient_object_store_error) — those recover on retry and aren't a defect, so reporting
+        them to error tracking is just noise. Never suppresses the re-raise itself, so Temporal's
+        activity retry policy is unaffected either way.
+        """
+        if is_transient_object_store_error(e):
+            await self._logger.awarning(f"get_delta_table: transient object-store error, not reporting: {e}")
+        else:
+            capture_exception(e)
+
     async def _evolve_delta_schema(self, schema: pa.Schema) -> deltalake.DeltaTable:
         delta_table = await self.get_delta_table()
         if delta_table is None:
@@ -310,7 +321,7 @@ async def get_delta_table(self) -> deltalake.DeltaTable | None:
             # best-effort maintenance to the main write path, so this can't safely swallow the
             # error and report "no table" here — that would trip should_overwrite_table for a
             # table that actually exists, risking data loss.
-            capture_exception(e)
+            await self._capture_unless_transient(e)
             raise
 
         if is_delta:
@@ -319,7 +330,7 @@ async def get_delta_table(self) -> deltalake.DeltaTable | None:
                     deltalake.DeltaTable, table_uri=delta_uri, storage_options=storage_options
                 )
             except Exception as e:
-                capture_exception(e)
+                await self._capture_unless_transient(e)
                 error_text = "".join(str(arg) for arg in e.args)
                 # Unrecoverable tables (bugged decimals, or an orphaned _delta_log missing its
                 # metadata action — impossible on a healthy table): wipe so the sync starts fresh.

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py

@@ -388,21 +388,44 @@ async def test_is_deltatable_failure_is_captured_and_reraised(self):
         helper = DeltaTableHelper(resource_name="t", job=MagicMock(), logger=_make_logger())
         delta_uri = "s3://bucket/team_id/job_id/t"
 
+        module = "products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper"
+        with (
+            patch.object(helper, "_get_delta_table_uri", AsyncMock(return_value=delta_uri)),
+            patch(f"{module}.deltalake.DeltaTable") as mock_delta_table,
+            patch(f"{module}.capture_exception") as mock_capture,
+        ):
+            mock_delta_table.is_deltatable.side_effect = OSError("Access Denied: not authorized to list bucket")
+
+            with pytest.raises(OSError, match="Access Denied"):
+                await helper.get_delta_table()
+
+            mock_capture.assert_called_once()
+            assert helper.is_first_sync is False
+
+    @pytest.mark.asyncio
+    async def test_is_deltatable_transient_error_is_not_captured_but_still_reraised(self):
+        """A known-transient object-store blip (e.g. an S3 LIST request timing out) must not be
+        reported to error tracking as a defect — it's a self-recovering network hiccup, not a bug —
+        but it must still propagate so Temporal's activity retry policy retries the sync."""
+        helper = DeltaTableHelper(resource_name="t", job=MagicMock(), logger=_make_logger())
+        delta_uri = "s3://bucket/team_id/job_id/t"
+
         module = "products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper"
         with (
             patch.object(helper, "_get_delta_table_uri", AsyncMock(return_value=delta_uri)),
             patch(f"{module}.deltalake.DeltaTable") as mock_delta_table,
             patch(f"{module}.capture_exception") as mock_capture,
         ):
             mock_delta_table.is_deltatable.side_effect = OSError(
-                "Generic S3 error: Received redirect without LOCATION, this normally indicates "
-                "an incorrectly configured region"
+                "Generic S3 error\nError getting list response body\nHTTP error\n"
+                "request or response body error\noperation timed out"
             )
 
-            with pytest.raises(OSError, match="Received redirect without LOCATION"):
+            with pytest.raises(OSError, match="operation timed out"):
                 await helper.get_delta_table()
 
-            mock_capture.assert_called_once()
+            mock_capture.assert_not_called()
+            cast(AsyncMock, helper._logger.awarning).assert_awaited_once()
             assert helper.is_first_sync is False