fix(data-imports): classify DB connection blips as transient in pre-write maintenance

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)

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:

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.