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..ed0160d41f66 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 @@ -182,7 +182,11 @@ async def _enabled_team_ids(self) -> list[int] | None: if self._team_ids_fetched_at is None: # Never had a set: claim nothing rather than everything. self._team_ids = [] - self._team_ids_fetched_at = now + # Always advance the timestamp, even on failure: otherwise a control-plane + # outage that starts after a successful resolution leaves it stuck in the + # past, and every poll tick (~2s) re-triggers this branch instead of the + # intended ENABLEMENT_REFRESH_SECONDS backoff. + self._team_ids_fetched_at = now return self._team_ids async def _run_maintenance(self, conn: psycopg.AsyncConnection[Any], team_ids: list[int] | None) -> None: 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..ce0dcf193f4a 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 @@ -16,6 +16,7 @@ PermanentBatchApplyError, ) from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer import ( + ENABLEMENT_REFRESH_SECONDS, DuckgresBatchConsumer, DuckgresBatchConsumerAdapter, DuckgresConsumerConfig, @@ -454,6 +455,29 @@ async def test_fetch_swallows_maintenance_query_timeout_and_skips_planner(self): mock_planner.assert_not_called() mock_fetch.assert_not_called() + @pytest.mark.asyncio + async def test_enablement_refresh_failure_after_success_waits_full_interval_before_retrying(self): + # A control-plane outage that starts after a prior successful resolution must + # back off for the full refresh interval, same as a fresh outage would — a + # stuck fetched_at previously made every ~2s poll tick retry immediately. + adapter = DuckgresBatchConsumerAdapter() + adapter._team_ids = [1, 2] + adapter._team_ids_fetched_at = time.monotonic() - ENABLEMENT_REFRESH_SECONDS - 1 + + with patch( + "products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.duckgres.consumer.database_sync_to_async_pool", + ) as mock_wrapper: + mock_wrapper.return_value = AsyncMock(side_effect=RuntimeError("duckgres control plane unreachable")) + + team_ids = await adapter._enabled_team_ids() + assert team_ids == [1, 2] + assert mock_wrapper.call_count == 1 + + # Polling again right away must not re-attempt the refresh. + team_ids_again = await adapter._enabled_team_ids() + assert team_ids_again == [1, 2] + assert mock_wrapper.call_count == 1 + class TestMidClaimRetire: @pytest.mark.asyncio