fix(data-imports): stop reporting duckgres sink maintenance timeouts as errors

products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/consumer.py

@@ -43,6 +43,7 @@
 )
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.jobs_db import (
     DuckgresBatchQueue,
+    is_eligibility_query_timeout,
 )
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.postgres_queue.jobs_db import (
     LEASE_TTL_SECONDS,
@@ -219,8 +220,14 @@ async def _run_maintenance(self, conn: psycopg.AsyncConnection[Any], team_ids: l
             )
             SINK_ORGS_AT_BUDGET.set(orgs_at_budget)
         except Exception as e:
-            logger.exception("duckgres_sink_maintenance_query_failed")
-            capture_exception(e)
+            if is_eligibility_query_timeout(e):
+                # Expected under a slow/loaded queue DB — the eligibility-CTE
+                # queries are timeout-bounded specifically so this fails fast
+                # and the next tick just retries; not worth an error-tracking report.
+                logger.warning("duckgres_sink_maintenance_query_timed_out")
+            else:
+                logger.exception("duckgres_sink_maintenance_query_failed")
+                capture_exception(e)
             return
 
         block_list_was_unset = self._blocked_schema_ids is None

products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/jobs_db.py

@@ -62,6 +62,13 @@ def _latest_status_lateral(status_table: str, batch_alias: str) -> str:
 ELIGIBILITY_QUERY_STATEMENT_TIMEOUT_MS = 30_000
 
 
+def is_eligibility_query_timeout(error: BaseException) -> bool:
+    """True when ``error`` is a query cancellation, the expected shape of
+    ELIGIBILITY_QUERY_STATEMENT_TIMEOUT_MS tripping under a slow/loaded queue DB.
+    Callers should skip the tick and retry rather than report it as a defect."""
+    return isinstance(error, psycopg.errors.QueryCanceled)
+
+
 @asynccontextmanager
 async def _statement_timeout(conn: psycopg.AsyncConnection[Any], timeout_ms: int) -> AsyncIterator[None]:
     """Bound the wrapped query with a server-side ``statement_timeout``.

products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_consumer.py

@@ -426,7 +426,9 @@ async def test_fetch_swallows_maintenance_query_timeout_and_skips_planner(self):
         # The eligibility-CTE maintenance queries are statement-timeout bounded;
         # a timeout (or any transient queue-DB error) must be swallowed so the
         # poll loop is neither wedged nor crashed — the planner is skipped and
-        # nothing is claimed this tick, and the next poll just retries.
+        # nothing is claimed this tick, and the next poll just retries. It's
+        # also the expected shape of that bound tripping under load, so it
+        # must not be reported to error tracking as a defect.
         adapter = DuckgresBatchConsumerAdapter()
         adapter._team_ids = [1, 2]
         adapter._team_ids_fetched_at = time.monotonic()
@@ -445,6 +447,45 @@ async def test_fetch_swallows_maintenance_query_timeout_and_skips_planner(self):
             patch(
                 "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer.run_backfill_planner",
             ) as mock_planner,
+            patch(
+                "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer.capture_exception",
+            ) as mock_capture,
+        ):
+            batches = await adapter.fetch_and_lock(
+                conn, limit=50, retry_backoff_base_seconds=0, owner_token="test-owner", lease_ttl_seconds=300
+            )
+
+        assert batches == []
+        mock_planner.assert_not_called()
+        mock_fetch.assert_not_called()
+        mock_capture.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_fetch_reports_unexpected_maintenance_query_error(self):
+        # A genuine, unexpected maintenance-query failure (not the queue DB's
+        # own statement timeout) must still be swallowed to protect the poll
+        # loop, but is worth surfacing to error tracking as a real defect.
+        adapter = DuckgresBatchConsumerAdapter()
+        adapter._team_ids = [1, 2]
+        adapter._team_ids_fetched_at = time.monotonic()
+        conn = _make_healthy_conn()
+
+        with (
+            patch(
+                "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer.DuckgresBatchQueue.get_delta_succeeded_and_lock",
+                new_callable=AsyncMock,
+            ) as mock_fetch,
+            patch(
+                "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer.DuckgresBatchQueue.supersede_replaced_runs",
+                new_callable=AsyncMock,
+                side_effect=RuntimeError("unexpected queue db failure"),
+            ),
+            patch(
+                "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer.run_backfill_planner",
+            ) as mock_planner,
+            patch(
+                "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer.capture_exception",
+            ) as mock_capture,
         ):
             batches = await adapter.fetch_and_lock(
                 conn, limit=50, retry_backoff_base_seconds=0, owner_token="test-owner", lease_ttl_seconds=300
@@ -453,6 +494,7 @@ async def test_fetch_swallows_maintenance_query_timeout_and_skips_planner(self):
         assert batches == []
         mock_planner.assert_not_called()
         mock_fetch.assert_not_called()
+        mock_capture.assert_called_once()
 
 
 class TestMidClaimRetire: