diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py index 432f03b5eaee..56918abdaaeb 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py +++ b/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). diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py index ffaa2208aacf..aec4417cf55c 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py +++ b/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() diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/pipeline.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/pipeline.py index 0a7e156c6005..47c508891a43 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/pipeline.py +++ b/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)}") diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/test/test_pipeline.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/test/test_pipeline.py index aa79fbdf7c0a..6ffb50d90f57 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v2/test/test_pipeline.py +++ b/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()