fix(data-imports): stop reporting transient S3 blips as compaction failures

products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py

@@ -294,6 +294,9 @@ async def run_post_load_operations(
     from products.warehouse_sources.backend.temporal.data_imports.pipelines.common.extract import (
         finalize_desc_sort_incremental_value,
     )
+    from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import (  # noqa: PLC0415 — keeps the heavy deltalake dep off this module's top-level import path
+        is_transient_object_store_error,
+    )
     from products.warehouse_sources.backend.temporal.data_imports.pipelines.helpers import build_table_name
     from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_sync import (
         register_cdc_companion_table,
@@ -346,16 +349,24 @@ async def run_post_load_operations(
                         schema.id, schema.team_id, updates={watermark_key: new_version}
                     )
         except Exception as e:
-            capture_exception(e)
-            logger.exception(f"Delta maintenance failed: {e}", exc_info=e)
+            if is_transient_object_store_error(e):
+                # A rate-limited or connectivity blip talking to our own S3 bucket isn't a bug - the
+                # next tick's maintenance pass retries the same idempotent cleanup.
+                logger.warning(f"Delta maintenance skipped: transient object-store error: {e}")
+            else:
+                capture_exception(e)
+                logger.exception(f"Delta maintenance failed: {e}", exc_info=e)
     else:
         logger.debug("Triggering compaction and vacuuming on delta table")
         try:
             with POST_LOAD_DURATION_SECONDS.labels(operation="compact").time():
                 await delta_table_helper.compact_table()
         except Exception as e:
-            capture_exception(e)
-            logger.exception(f"Compaction failed: {e}", exc_info=e)
+            if is_transient_object_store_error(e):
+                logger.warning(f"Compaction skipped: transient object-store error: {e}")
+            else:
+                capture_exception(e)
+                logger.exception(f"Compaction failed: {e}", exc_info=e)
 
     if is_cdc_companion:
         # Look up the existing companion table's queryable_folder (not the main schema.table).

products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py

@@ -174,18 +174,47 @@ async def test_watermark_persistence(self, _name: str, returned_version: int | N
         else:
             update_config.assert_not_called()
 
+    @parameterized.expand(
+        [
+            # A genuine maintenance bug must still be captured for visibility.
+            ("genuine_bug", RuntimeError("maintenance blew up"), True),
+            # A transient S3 rate-limit/connectivity blip is already non-fatal here (the next
+            # tick's maintenance retries the same idempotent cleanup) and must not be promoted
+            # into a fresh error-tracking issue — the regression this guards.
+            ("transient_s3_slowdown", OSError("Generic S3 error: Please reduce your request rate."), False),
+        ]
+    )
     @pytest.mark.asyncio
-    async def test_maintenance_failure_does_not_fail_post_load(self):
-        # A maintenance hiccup (S3 flake) must not fail the final batch — the rest of post-load
+    async def test_maintenance_failure_handling(self, _name: str, error: Exception, expect_capture: bool):
+        # A maintenance hiccup must not fail the final batch — the rest of post-load
         # (queryable folder prep, table registration) still has to run or the job wedges.
         schema = _make_schema(is_cdc=True)
         helper = _make_helper()
-        helper.run_maintenance = AsyncMock(side_effect=RuntimeError("maintenance blew up"))
+        helper.run_maintenance = AsyncMock(side_effect=error)
 
         with patch(f"{_LOAD_MODULE}.capture_exception") as mock_capture:
             _, prepare_s3 = await _run_post_load(schema, helper, cdc_write_mode="incremental")
 
-        mock_capture.assert_called_once()
+        assert mock_capture.called is expect_capture
+        prepare_s3.assert_awaited_once()
+
+    @parameterized.expand(
+        [
+            ("genuine_bug", RuntimeError("compaction blew up"), True),
+            ("transient_s3_slowdown", OSError("Generic S3 error: Please reduce your request rate."), False),
+        ]
+    )
+    @pytest.mark.asyncio
+    async def test_compact_failure_handling(self, _name: str, error: Exception, expect_capture: bool):
+        # Same non-fatal handling as maintenance, for the non-CDC unconditional compact_table path.
+        schema = _make_schema(is_cdc=False)
+        helper = _make_helper()
+        helper.compact_table = AsyncMock(side_effect=error)
+
+        with patch(f"{_LOAD_MODULE}.capture_exception") as mock_capture:
+            _, prepare_s3 = await _run_post_load(schema, helper)
+
+        assert mock_capture.called is expect_capture
         prepare_s3.assert_awaited_once()
 
 

products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/pipeline.py

@@ -54,7 +54,10 @@
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.async_iterate import async_iterate
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.batcher import Batcher
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.cdp_producer import CDPProducer
-from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import DeltaTableHelper
+from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import (
+    DeltaTableHelper,
+    is_transient_object_store_error,
+)
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.hogql_schema import HogQLSchema
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.partitioning import setup_partitioning
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.person_property_row_sink import (
@@ -454,8 +457,13 @@ async def _post_run_operations(self, row_count: int) -> str | None:
         try:
             await self._delta_table_helper.compact_table()
         except Exception as e:
-            capture_exception(e)
-            await self._logger.aexception(f"Compaction failed: {e}", exc_info=e)
+            if is_transient_object_store_error(e):
+                # A rate-limited or connectivity blip compacting/vacuuming our own S3 bucket isn't a
+                # bug - the next sync's post-run compaction retries the same idempotent cleanup.
+                await self._logger.awarning(f"Compaction skipped: transient object-store error: {e}")
+            else:
+                capture_exception(e)
+                await self._logger.aexception(f"Compaction failed: {e}", exc_info=e)
 
         file_uris = await self._delta_table_helper.get_file_uris()
         await self._logger.adebug(f"Preparing S3 files - total parquet files: {len(file_uris)}")

products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/test/test_pipeline.py

@@ -1,12 +1,18 @@
+import uuid
 from typing import cast
 
 import pytest
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
+from parameterized import parameterized
+
+from products.warehouse_sources.backend.models.external_data_schema import ExternalDataSchema
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.cdp_producer import CDPProducer
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v2.pipeline import PipelineNonDLT
 from products.warehouse_sources.backend.temporal.data_imports.sources.common.typings import SourceResponse
 
+_PIPELINE_MODULE = "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v2.pipeline"
+
 
 @pytest.mark.asyncio
 async def test_run_cleanup_failure_does_not_mask_import_error(monkeypatch):
@@ -38,3 +44,66 @@ async def _raise_import_error(_cdp_producer):
         await pipeline.run()
 
     pipeline._logger.aexception.assert_awaited_once_with("Failed to clean up delta table helper")
+
+
+def _make_pipeline_for_post_run_operations() -> PipelineNonDLT:
+    pipeline = PipelineNonDLT.__new__(PipelineNonDLT)
+    pipeline._logger = AsyncMock()
+    pipeline._delta_table_helper = AsyncMock()
+    pipeline._delta_table_helper.get_delta_table.return_value = MagicMock()
+    pipeline._delta_table_helper.get_file_uris.return_value = []
+    pipeline._job = MagicMock(id=uuid.uuid4(), team_id=1)
+    pipeline._table = None
+    pipeline._resource_name = "orders"
+    pipeline._schema = MagicMock()
+    pipeline._schema.initial_sync_complete = True
+    # Non-CDC skips the CDC-companion-seeding block, so only validate_schema_and_update_table
+    # (mocked below) needs to be reachable.
+    pipeline._schema.sync_type = ExternalDataSchema.SyncType.INCREMENTAL
+    pipeline._schema.cdc_table_mode = "consolidated"
+    pipeline._source = MagicMock()
+    pipeline._resource = MagicMock()
+    pipeline._internal_schema = MagicMock()
+    pipeline._last_incremental_field_value = None
+    return pipeline
+
+
+def _fake_database_sync_to_async_pool(fn):
+    async def _inner(*args, **kwargs):
+        return fn(*args, **kwargs)
+
+    return _inner
+
+
+class TestPostRunOperationsCompactionErrorHandling:
+    # Regression: `_post_run_operations` used to call `capture_exception` for every compaction
+    # failure, including transient S3 rate-limit/connectivity blips that are already non-fatal
+    # (the sync completes regardless) — flooding error tracking with noise. A genuine compaction
+    # bug must still be captured.
+    @parameterized.expand(
+        [
+            ("transient_s3_slowdown", OSError("Generic S3 error: Please reduce your request rate."), False),
+            ("genuine_bug", OSError("Access Denied: not authorized"), True),
+        ]
+    )
+    @pytest.mark.asyncio
+    async def test_compaction_error_handling(self, _name: str, error: Exception, expect_capture: bool):
+        pipeline = _make_pipeline_for_post_run_operations()
+        cast(AsyncMock, pipeline._delta_table_helper.compact_table).side_effect = error
+
+        with (
+            patch(f"{_PIPELINE_MODULE}.capture_exception") as mock_capture,
+            patch(f"{_PIPELINE_MODULE}.database_sync_to_async_pool", _fake_database_sync_to_async_pool),
+            patch(f"{_PIPELINE_MODULE}.prepare_s3_files_for_querying", AsyncMock(return_value="orders__query_1")),
+            patch(f"{_PIPELINE_MODULE}.update_last_synced_at", AsyncMock()),
+            patch(f"{_PIPELINE_MODULE}.notify_revenue_analytics_that_sync_has_completed", AsyncMock()),
+            patch(f"{_PIPELINE_MODULE}.finalize_desc_sort_incremental_value", AsyncMock()),
+            patch(f"{_PIPELINE_MODULE}.validate_schema_and_update_table", AsyncMock()),
+        ):
+            await pipeline._post_run_operations(row_count=10)
+
+        assert mock_capture.called is expect_capture
+        if expect_capture:
+            cast(AsyncMock, pipeline._logger.aexception).assert_awaited_once()
+        else:
+            cast(AsyncMock, pipeline._logger.awarning).assert_awaited_once()