From ed23b381f18da0d03abcfb8b98e614b04b005c9e Mon Sep 17 00:00:00 2001 From: Ulysse Pence Date: Wed, 12 Aug 2026 15:01:08 -0200 Subject: [PATCH] fix(deriver): increase deriver polling backoff --- src/config.py | 8 +-- src/deriver/queue_manager.py | 23 ++++++- tests/test_db_resilience.py | 125 +++++++++++++++++++++++++++++++++++ 3 files changed, 150 insertions(+), 6 deletions(-) diff --git a/src/config.py b/src/config.py index 34a5917b..f3595f0b 100644 --- a/src/config.py +++ b/src/config.py @@ -849,8 +849,8 @@ class DeriverSettings(HonchoSettings): # Reduces steady-state query load against the (shared) DB/pooler. POLLING_BACKOFF_ENABLED: bool = True POLLING_SLEEP_MAX_INTERVAL_SECONDS: Annotated[ - float, Field(default=30.0, gt=0.0, le=300.0) - ] = 30.0 + float, Field(default=120.0, gt=0.0, le=300.0) + ] = 120.0 POLLING_BACKOFF_MULTIPLIER: Annotated[ float, Field(default=2.0, ge=1.0, le=10.0) ] = 2.0 @@ -861,10 +861,10 @@ class DeriverSettings(HonchoSettings): float, Field(default=30.0, ge=0.0, le=300.0) ] = 30.0 # Multiply every poll sleep by a random factor in [1 - ratio, 1 + ratio] - # (0.5 -> [0.5x, 1.5x]) so poll loops don't re-converge over time. The + # (0.25 -> [0.75x, 1.25x]) so poll loops don't re-converge over time. The # backoff schedule is unchanged; only the returned sleep is scattered. Set # to 0.0 to disable. - POLLING_JITTER_RATIO: Annotated[float, Field(default=0.5, ge=0.0, le=1.0)] = 0.5 + POLLING_JITTER_RATIO: Annotated[float, Field(default=0.25, ge=0.0, le=1.0)] = 0.25 STALE_SESSION_TIMEOUT_MINUTES: Annotated[int, Field(default=5, gt=0, le=1440)] = 5 # Minimum (jittered) spacing between stale-work-unit cleanup runs STALE_WORK_UNIT_CLEANUP_INTERVAL_SECONDS: Annotated[ diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 498372ed..5c1a7fe1 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -4,7 +4,7 @@ import random import signal import time from asyncio import Task -from collections.abc import Sequence +from collections.abc import Iterable, Sequence from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from logging import getLogger @@ -476,6 +476,24 @@ class QueueManager: """Snap the polling interval back to the base after finding work.""" self._current_poll_interval = settings.DERIVER.POLLING_SLEEP_INTERVAL_SECONDS + @staticmethod + def _is_tenant_work(work_unit_keys: Iterable[str]) -> bool: + """True if any claimed work unit is real tenant work, not housekeeping. + + The reconciler enqueues its own sweeps on a fixed timer, so treating them + as "the queue is busy" would reset the backoff every cycle and keep the + deriver polling often enough that the pooler never releases an idle + tenant's connection. Unparseable keys count as tenant work so an unknown + key can never strand the loop in a long sleep. + """ + for key in work_unit_keys: + try: + if parse_work_unit_key(key).task_type != "reconciler": + return True + except ValueError: + return True + return False + def _jitter(self, seconds: float) -> float: """Scatter a sleep by +/- POLLING_JITTER_RATIO to avoid lockstep polling. @@ -542,7 +560,8 @@ class QueueManager: await self._maybe_cleanup_stale_work_units() claimed_work_units = await self.get_and_claim_work_units() if claimed_work_units: - self._reset_poll_interval() + if self._is_tenant_work(claimed_work_units): + self._reset_poll_interval() for work_unit_key, aqs_id in claimed_work_units.items(): # Create a new task for processing this work unit if not self.shutdown_event.is_set(): diff --git a/tests/test_db_resilience.py b/tests/test_db_resilience.py index faffa89a..76b2d0d4 100644 --- a/tests/test_db_resilience.py +++ b/tests/test_db_resilience.py @@ -175,6 +175,131 @@ async def test_polling_loop_idle_sleeps_once_per_cycle( assert sleeps == [1.0, 2.0, 4.0, 8.0, 8.0] +def test_is_tenant_work_ignores_reconciler_only_batches() -> None: + from src.deriver.queue_manager import QueueManager + + assert not QueueManager._is_tenant_work(["reconciler:sync_vectors"]) # pyright: ignore[reportPrivateUsage] + assert not QueueManager._is_tenant_work( # pyright: ignore[reportPrivateUsage] + ["reconciler:sync_vectors", "reconciler:cleanup_queue"] + ) + # A mixed batch is tenant work: real work is present alongside housekeeping. + assert QueueManager._is_tenant_work( # pyright: ignore[reportPrivateUsage] + ["reconciler:sync_vectors", "representation:ws:sess:peer"] + ) + # Unparseable keys count as tenant work so an unknown key can't strand the + # loop in a long sleep. + assert QueueManager._is_tenant_work(["not-a-real-key"]) # pyright: ignore[reportPrivateUsage] + + +@pytest.mark.asyncio +async def test_reconciler_work_does_not_reset_backoff( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The reconciler enqueues sweeps on its own timer. Claiming one must not + look like a busy queue, or the backoff resets every cycle and the pooler + never releases an idle tenant's connection.""" + monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_ENABLED", True) + monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_INTERVAL_SECONDS", 1.0) + monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_MULTIPLIER", 2.0) + monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_MAX_INTERVAL_SECONDS", 64.0) + monkeypatch.setattr(settings.DERIVER, "POLLING_JITTER_RATIO", 0.0) + + import asyncio + + from src.deriver import queue_manager as qm_mod + + qm = qm_mod.QueueManager() + sleeps: list[float] = [] + polls = {"n": 0} + + async def fake_cleanup() -> None: + return None + + async def fake_claim() -> dict[str, str]: + polls["n"] += 1 + if polls["n"] >= 5: + qm.shutdown_event.set() + # Third poll hands back a reconciler sweep; the rest are empty. + return {"reconciler:sync_vectors": "aqs-1"} if polls["n"] == 3 else {} + + async def fake_process(_work_unit_key: str, _worker_id: str) -> None: + return None + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + + monkeypatch.setattr(qm, "cleanup_stale_work_units", fake_cleanup) + monkeypatch.setattr(qm, "get_and_claim_work_units", fake_claim) + monkeypatch.setattr(qm, "process_work_unit", fake_process) + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + await qm.polling_loop() + + # Polls 1 and 2 sleep 1 and 2. Poll 3 claims the sweep, so it neither sleeps + # nor resets. Polls 4 and 5 resume the schedule at 4 -- not back at 1. + assert sleeps == [1.0, 2.0, 4.0, 8.0] + + +@pytest.mark.asyncio +async def test_tenant_work_still_resets_backoff( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Real tenant work must still snap the interval back for fast pickup.""" + monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_ENABLED", True) + monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_INTERVAL_SECONDS", 1.0) + monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_MULTIPLIER", 2.0) + monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_MAX_INTERVAL_SECONDS", 64.0) + monkeypatch.setattr(settings.DERIVER, "POLLING_JITTER_RATIO", 0.0) + + import asyncio + + from src.deriver import queue_manager as qm_mod + + qm = qm_mod.QueueManager() + sleeps: list[float] = [] + polls = {"n": 0} + + async def fake_cleanup() -> None: + return None + + async def fake_claim() -> dict[str, str]: + polls["n"] += 1 + if polls["n"] >= 5: + qm.shutdown_event.set() + return {"representation:ws:sess:peer": "aqs-1"} if polls["n"] == 3 else {} + + async def fake_process(_work_unit_key: str, _worker_id: str) -> None: + return None + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + + monkeypatch.setattr(qm, "cleanup_stale_work_units", fake_cleanup) + monkeypatch.setattr(qm, "get_and_claim_work_units", fake_claim) + monkeypatch.setattr(qm, "process_work_unit", fake_process) + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + await qm.polling_loop() + + # Poll 3 finds tenant work, so polls 4 and 5 start over from the base + # interval rather than continuing from 4. + assert sleeps == [1.0, 2.0, 1.0, 2.0] + + +def test_backoff_cap_clears_pooler_idle_timeout() -> None: + """Guard the mechanism this backoff exists for. + + An idle tenant's connection is released only when the AlloyDB managed + pooler's server_connection_idle_timeout elapses between polls, so the + SHORTEST jittered sleep at the cap -- not the average -- has to clear it. + If someone raises the jitter ratio or lowers the cap, this fails loudly. + """ + pooler_idle_timeout_seconds = 45.0 + cap = settings.DERIVER.POLLING_SLEEP_MAX_INTERVAL_SECONDS + shortest_sleep_at_cap = cap * (1.0 - settings.DERIVER.POLLING_JITTER_RATIO) + assert shortest_sleep_at_cap > pooler_idle_timeout_seconds + + def test_inflight_gauge_no_drift(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(settings.METRICS, "NAMESPACE", "test") child: Any = db_queries_in_flight_gauge.labels(instance_type="api")