From 3a9e27e7c629ec321914b0ff61c1229bcc09a547 Mon Sep 17 00:00:00 2001 From: Tom Owers Date: Fri, 24 Jul 2026 16:38:45 +0100 Subject: [PATCH] fix(data-imports): stop reporting duckgres sink maintenance timeouts as errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The eligibility-CTE maintenance queries (`supersede_replaced_runs`, `get_backlog_stats`, `count_orgs_at_budget`) are bounded by a 30s statement timeout specifically so a slow/loaded queue DB fails fast and the poll loop retries on the next tick — the surrounding code already documents this as expected. But the catch-all in `_run_maintenance` reported every exception, including that expected timeout, to error tracking as a defect. Added `is_eligibility_query_timeout` next to the timeout constant in `jobs_db.py` and branched on it in `consumer.py`: a `QueryCanceled` from these queries now logs a warning and skips the tick as before, without `capture_exception`; any other, genuinely unexpected error still gets captured. Generated-By: PostHog Code Task-Id: 74c8471e-3e63-4af7-850f-5b69730bc85b --- .../pipeline_v3/duckgres/consumer.py | 11 ++++- .../pipelines/pipeline_v3/duckgres/jobs_db.py | 7 +++ .../pipeline_v3/duckgres/test_consumer.py | 44 ++++++++++++++++++- 3 files changed, 59 insertions(+), 3 deletions(-) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/consumer.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/consumer.py index f789af1b6ed8..24e875c19e99 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/consumer.py +++ b/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 diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/jobs_db.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/jobs_db.py index 1e8aa24cf1c0..2454e9131f39 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/jobs_db.py +++ b/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``. diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_consumer.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_consumer.py index 81ccbd863bf4..48c6fb0cc378 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_consumer.py +++ b/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: