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 0100f9426530..c90d80959fb3 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 @@ -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. diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py index 3eabefb9e63a..02fb73a09c0f 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py @@ -388,6 +388,28 @@ 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)), @@ -395,14 +417,15 @@ async def test_is_deltatable_failure_is_captured_and_reraised(self): 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