Stagger periodic poll sweeps to avoid thundering herd

All pollers sharing the same poll_interval_s previously woke up and
swept at exactly the same moment after startup. With 10+ sources
this causes a coordinated spike in listing traffic (S3 LIST, WebDAV
PROPFIND, HTTP HEAD) every interval.

Add a random initial delay of 0-25% of the poll interval after the
first sweep, applied to both PeriodicPoller and FSPoller's sweep
loop. Subsequent sweeps run on the normal fixed interval, now
staggered across sources.

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:

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: