diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py b/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py index d619e8e1..6bb714a2 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py @@ -1,5 +1,6 @@ import asyncio import logging +import random from pathlib import Path from typing import TYPE_CHECKING @@ -64,10 +65,19 @@ class FSPoller(BasePoller): """Periodic full sweep. Catches files modified while the watcher wasn't running (gaps between starts, races, FS events the OS dropped). Sweep behaviour is unit-tested via `_sweep_once()` directly.""" + # Stagger the first sleep so multiple FS pollers don't all sweep + # at exactly the same moment after startup. + interval = self.config.poll_interval_s + jitter = random.uniform(0, interval * 0.25) + try: + await asyncio.wait_for(self._stop.wait(), timeout=jitter) + return + except TimeoutError: + pass while not self._stop.is_set(): try: await asyncio.wait_for( - self._stop.wait(), timeout=self.config.poll_interval_s + self._stop.wait(), timeout=interval ) return except TimeoutError: diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py b/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py index a19c233b..712e2ce2 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py @@ -1,5 +1,6 @@ import asyncio import logging +import random from typing import TYPE_CHECKING from haiku.rag.ingester.pollers.base import BasePoller @@ -38,10 +39,19 @@ class PeriodicPoller(BasePoller): # immediately instead of waiting one full interval. The sweep # behaviour itself is exercised via `_sweep_once()` unit tests. await self._sweep_once() + # Stagger the first sleep so pollers that share the same interval + # don't all wake up and sweep at exactly the same moment. + interval = self.config.poll_interval_s + jitter = random.uniform(0, interval * 0.25) + try: + await asyncio.wait_for(self._stop.wait(), timeout=jitter) + return + except TimeoutError: + pass while not self._stop.is_set(): try: await asyncio.wait_for( - self._stop.wait(), timeout=self.config.poll_interval_s + self._stop.wait(), timeout=interval ) return except TimeoutError: