From 1a9f8eaee1c312cf19034a2dffd7a86c6232e2cc Mon Sep 17 00:00:00 2001 From: Tom Owers Date: Thu, 30 Jul 2026 20:57:53 +0100 Subject: [PATCH] fix(data-imports): classify DB connection blips as transient in pre-write maintenance Pre-write maintenance (run_pre_write_defensive_compact) can hit an app-DB connection blip (DNS resolution failure, pooler-dropped connection) while resolving job.folder_path() on a pooled connection. Those surface as OperationalError/InterfaceError, are self-healing on the next sync's maintenance pass, and shouldn't be reported to error tracking as maintenance bugs. Add is_transient_maintenance_error, which broadens the existing object-store classification to also cover these DB connection blips, matching the OperationalError/InterfaceError classification already used by repartition_table.py's _is_transient_infra_error. The maintenance path uses the broadened check; is_transient_object_store_error is retained unchanged for the object-store-specific callers. --- .../data_imports/pipelines/common/extract.py | 14 +++++----- .../pipelines/common/test/test_extract.py | 26 +++++++++++++++++++ .../pipelines/core/delta_table_helper.py | 15 +++++++++++ 3 files changed, 48 insertions(+), 7 deletions(-) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py index 9ef2d758212e..22274f60e4a7 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py @@ -21,7 +21,7 @@ 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, - is_transient_delta_maintenance_error, + is_transient_maintenance_error, is_transient_object_store_error, ) from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.person_property_row_sink import ( @@ -705,10 +705,10 @@ async def run_pre_write_defensive_compact( the CDC post-load path in `common/load.py` writes the same watermark, and both merge via `update_sync_type_config_keys` under a row lock. Wrapped in try/except so a maintenance failure never blocks the actual sync; the original error path is unaffected. - A transient object-store error (see `is_transient_object_store_error`) or a racy - concurrent-maintenance DeltaError (see `is_transient_delta_maintenance_error`) is - logged at warning instead of captured — the next sync's maintenance pass retries it - from scratch, so neither is a bug in this function. + A transient infra error (see `is_transient_maintenance_error`) — an object-store hiccup, a racy + concurrent-maintenance DeltaError, or an app-DB connection blip — is logged at warning instead of + captured. The next sync's maintenance pass retries it from scratch, so it isn't a bug in this + function, just a temporary blip talking to our own S3 bucket, delta table, or app DB. Used by both `PipelineNonDLT.run` (v2) and `PipelineV3.run` to keep the behaviour identical across pipelines without each having to know how to look up `partition_count` @@ -732,8 +732,8 @@ async def run_pre_write_defensive_compact( schema.id, schema.team_id, updates={"last_vacuum_version": new_version} ) except Exception as e: - if is_transient_object_store_error(e) or is_transient_delta_maintenance_error(e): - await logger.awarning(f"Pre-write maintenance skipped: transient error: {e}") + if is_transient_maintenance_error(e): + await logger.awarning(f"Pre-write maintenance skipped: transient infra error: {e}") return capture_exception(e) await logger.aexception(f"Pre-write maintenance failed: {e}", exc_info=e) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py index c8d629d4eca9..ed7ef506270a 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py @@ -5,6 +5,8 @@ from posthog.test.base import BaseTest from unittest.mock import AsyncMock, MagicMock, patch +from django.db import InterfaceError, OperationalError + import deltalake.exceptions from asgiref.sync import async_to_sync from parameterized import parameterized @@ -235,6 +237,30 @@ async def test_logs_transient_delta_maintenance_race_without_capturing(self): logger.awarning.assert_awaited_once() logger.aexception.assert_not_awaited() + @parameterized.expand( + [ + ("dns_resolution_failure", OperationalError, "[Errno -2] Name or service not known"), + ("pooler_dropped_connection", InterfaceError, "connection already closed"), + ] + ) + @pytest.mark.asyncio + async def test_logs_transient_db_connection_error_without_capturing( + self, _name: str, error_cls: type[Exception], error_message: str + ): + # A DNS/pooler blip hit while resolving `job.folder_path()` on a pooled app-DB connection + # (e.g. inside `_get_delta_table_uri`) isn't a maintenance bug either — same treatment as + # the object-store blips above, so a self-healing retry doesn't flood error tracking. + 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={}) + with patch(f"{_EXTRACT_MODULE}.capture_exception") as mock_capture: + await run_pre_write_defensive_compact(helper, schema, MagicMock(partition_count=None), logger) + + mock_capture.assert_not_called() + logger.awarning.assert_awaited_once() + logger.aexception.assert_not_awaited() + class TestReportHeartbeatTimeoutRecording(BaseTest): def _schema(self) -> ExternalDataSchema: 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 6027d575b141..5be95cf41c43 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 @@ -4,6 +4,7 @@ from typing import Any, Literal from django.conf import settings +from django.db import InterfaceError, OperationalError import numpy as np import pyarrow as pa @@ -109,6 +110,20 @@ def is_transient_delta_maintenance_error(error: BaseException) -> bool: _PURGE_S3_PREFIX_MAX_ATTEMPTS = 4 +def is_transient_maintenance_error(error: BaseException) -> bool: + """Infra blips seen during pre-write maintenance that aren't a maintenance bug. + + Covers S3/object-store hiccups reaching our own data-warehouse bucket (see + `is_transient_object_store_error` above), racy concurrent-maintenance DeltaErrors (see + `is_transient_delta_maintenance_error` above), and app-DB connection blips (DNS, pooler drops) hit + while resolving `job.folder_path()` on a pooled connection — the same `OperationalError`/`InterfaceError` + classification used for this failure class in `repartition_table.py`'s `_is_transient_infra_error`. + """ + if isinstance(error, OperationalError | InterfaceError): + return True + return is_transient_object_store_error(error) or is_transient_delta_maintenance_error(error) + + def _delta_merge_spill_kwargs() -> dict[str, int]: """delta-rs `merge` kwargs that let DataFusion spill to disk instead of OOMing on large merges.