From 76dfc95c562ee63505944bc3b852ec3eb3a69253 Mon Sep 17 00:00:00 2001
From: Tom Owers <owerstom@gmail.com>
Date: Mon, 27 Jul 2026 17:14:09 +0100
Subject: [PATCH] fix(data-imports): back off the full interval after a
 duckgres enablement refresh failure

The duckgres sink's `_enabled_team_ids` cache only advanced its `_team_ids_fetched_at` timestamp on the very first refresh failure. Once a refresh had succeeded at least once, a later failure left the timestamp stuck in the past, so every subsequent poll tick (~2s) re-attempted the refresh instead of waiting the intended `ENABLEMENT_REFRESH_SECONDS` (60s) window. Now the timestamp always advances on failure, so a transient control-plane outage backs off at the same cadence as a healthy refresh.

Generated-By: PostHog Code
Task-Id: cd517bae-419a-451a-87b1-f0edaf1c0516
---
 .../pipeline_v3/duckgres/consumer.py          |  6 ++++-
 .../pipeline_v3/duckgres/test_consumer.py     | 24 +++++++++++++++++++
 2 files changed, 29 insertions(+), 1 deletion(-)

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