fix: Drive repeating tasks from a delay source

ldclient/impl/aio/concurrency.py

@@ -9,11 +9,11 @@
 
 import asyncio
 import inspect
-import time
 from queue import Empty as QueueEmpty  # noqa: F401  (shared timeout exception)
 from queue import Full as QueueFull  # noqa: F401  (shared capacity exception)
 from typing import Any, Callable, Coroutine, Optional, Set
 
+from ldclient.impl.delay import DelaySource, FixedDelay
 from ldclient.impl.util import log
 
 
@@ -189,33 +189,42 @@ async def stop_all(self, timeout: float = 1) -> None:
 
 
 class AsyncRepeatingTask:
-    """Calls a callback repeatedly at fixed intervals on a background task.
+    """Calls a callback repeatedly on a background task, waiting whatever its
+    :class:`~ldclient.impl.delay.DelaySource` gives.
     Mirrors the semantics of ``ldclient.impl.repeating_task.RepeatingTask``:
-    the interval is measured from the start of each invocation, exceptions
-    from the callback are logged, and ``stop()`` prevents any further
-    invocations but cannot be undone."""
+    the wait starts when the callback returns, exceptions from the callback
+    are logged, and ``stop()`` prevents any further invocations but cannot be
+    undone."""
 
-    def __init__(self, label: str, interval: float, initial_delay: float, callable: Callable):
+    def __init__(self, label: str, delays: DelaySource, initial_delay: float, callable: Callable):
         self.__label = label
-        self.__interval = interval
+        self.__delays = delays
         self.__initial_delay = initial_delay
         self.__action = callable
         self.__stop = AsyncEvent()
         self.__task: Optional[asyncio.Task] = None
 
+    @staticmethod
+    def at_interval(label: str, interval: float, initial_delay: float, callable: Callable) -> 'AsyncRepeatingTask':
+        """Creates a task that runs at a fixed interval."""
+        return AsyncRepeatingTask(label, FixedDelay(interval), initial_delay, callable)
+
     def start(self):
-        """Starts the background task. Like a thread, the task can only be
-        started once."""
+        """Starts the background task, if it is not running already."""
         if self.__task is not None:
-            raise RuntimeError("tasks can only be started once")
+            log.info("Task %s has already been started; ignoring" % self.__label)
+            return
         self.__task = asyncio.ensure_future(self._run())
+        self.__task.add_done_callback(_log_task_exception)
         try:
             self.__task.set_name(f"{self.__label}.repeating")
         except AttributeError:
             pass
 
     def stop(self):
-        """Tells the background task to stop. It cannot be restarted after this."""
+        """Tells the background task to stop.
+
+        The stop is permanent. A later ``start()`` does not resume the task."""
         self.__stop.set()
         task = self.__task
         # When stop() is called from within the action itself, let the loop
@@ -237,14 +246,15 @@ async def _run(self):
                     return
             stopped = self.__stop.is_set()
             while not stopped:
-                next_time = time.time() + self.__interval
                 try:
                     result = self.__action()
                     if inspect.isawaitable(result):
                         await result
                 except Exception as e:
                     log.exception("Unexpected exception on worker task: %s" % e)
-                delay = next_time - time.time()
+                # The wait starts when the callback returns, so a slow callback
+                # never shortens it.
+                delay = self.__delays.next_delay
                 if delay > 0:
                     stopped = await self.__stop.wait(delay)
                 else:

ldclient/impl/async_big_segments.py

@@ -65,7 +65,7 @@ def __init__(self, config: AsyncBigSegmentsConfig):
 
         if self.__store:
             self.__cache = ExpiringDict(max_len=config.context_cache_size, max_age_seconds=config.context_cache_time)
-            self.__poll_task = AsyncRepeatingTask("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status)
+            self.__poll_task = AsyncRepeatingTask.at_interval("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status)
 
     def start(self):
         """Starts the status polling task. Separated from __init__ so the manager

ldclient/impl/big_segments.py

@@ -67,7 +67,7 @@ def __init__(self, config: BigSegmentsConfig):
 
         if self.__store:
             self.__cache = ExpiringDict(max_len=config.context_cache_size, max_age_seconds=config.context_cache_time)
-            self.__poll_task = RepeatingTask("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status)
+            self.__poll_task = RepeatingTask.at_interval("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status)
             self.__poll_task.start()
 
     def stop(self):

ldclient/impl/datasource/async_polling.py

@@ -33,7 +33,7 @@ def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequester, store:
         self._requester = requester
         self._store = store
         self._ready = ready
-        self._task = AsyncRepeatingTask("ldclient.datasource.polling", config.poll_interval, 0, self._fetch_and_store)
+        self._task = AsyncRepeatingTask.at_interval("ldclient.datasource.polling", config.poll_interval, 0, self._fetch_and_store)
 
     def start(self):
         log.info("Starting AsyncPollingUpdateProcessor with request interval: " + str(self._config.poll_interval))

ldclient/impl/datasource/polling.py

@@ -44,7 +44,7 @@ def __init__(self, config: Config, requester: FeatureRequester, store: FeatureSt
         self._requester = requester
         self._store = store
         self._ready = ready
-        self._task = RepeatingTask("ldclient.datasource.polling", config.poll_interval, 0, self._poll)
+        self._task = RepeatingTask.at_interval("ldclient.datasource.polling", config.poll_interval, 0, self._poll)
 
     def start(self):
         log.info("Starting PollingUpdateProcessor with request interval: " + str(self._config.poll_interval))

ldclient/impl/datasystem/async_fdv2.py

@@ -148,7 +148,7 @@ def _update_availability(self, available: bool) -> None:
         else:
             log.warning("Detected persistent store unavailability; updates will be cached until it recovers")
             if self._poller is None:
-                task_to_start = AsyncRepeatingTask("ldclient.check-availability", 0.5, 0, self._check_availability)
+                task_to_start = AsyncRepeatingTask.at_interval("ldclient.check-availability", 0.5, 0, self._check_availability)
                 self._poller = task_to_start
 
         self._status_sink(DataStoreStatus(available, True))
@@ -545,7 +545,7 @@ async def _consume_synchronizer_results(
         :return: the ConditionDirective describing how to proceed
         """
         action_queue: AsyncQueue = AsyncQueue()
-        timer = AsyncRepeatingTask(
+        timer = AsyncRepeatingTask.at_interval(
             label="AsyncFDv2-sync-cond-timer",
             interval=10,
             initial_delay=10,

ldclient/impl/datasystem/fdv1.py

@@ -101,7 +101,7 @@ def __update_availability(self, available: bool):
             return
 
         log.warn("Detected persistent store unavailability; updates will be cached until it recovers")
-        task = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability)
+        task = RepeatingTask.at_interval("ldclient.check-availability", 0.5, 0, self.__check_availability)
 
         with self.__lock.write():
             self.__poller = task

ldclient/impl/datasystem/fdv2.py

@@ -129,7 +129,7 @@ def __update_availability(self, available: bool):
                 poller_to_stop = self.__poller
                 self.__poller = None
             elif self.__poller is None:
-                task_to_start = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability)
+                task_to_start = RepeatingTask.at_interval("ldclient.check-availability", 0.5, 0, self.__check_availability)
                 self.__poller = task_to_start
 
         if available:
@@ -536,7 +536,7 @@ def _consume_synchronizer_results(
         :return: the ConditionDirective describing how to proceed
         """
         action_queue: Queue = Queue()
-        timer = RepeatingTask(
+        timer = RepeatingTask.at_interval(
             label="FDv2-sync-cond-timer",
             interval=10,
             initial_delay=10,

ldclient/impl/delay.py

@@ -0,0 +1,28 @@
+"""
+The wait a repeating task takes between invocations. Both schedulers read it,
+so it belongs to neither.
+"""
+
+# currently excluded from documentation - see docs/README.md
+
+from typing import Protocol
+
+
+class DelaySource(Protocol):
+    """Supplies the wait before a repeating task's next invocation."""
+
+    @property
+    def next_delay(self) -> float:
+        """The seconds to wait before the next invocation."""
+        ...
+
+
+class FixedDelay(DelaySource):
+    """A :class:`DelaySource` that always gives the same wait."""
+
+    def __init__(self, seconds: float):
+        self.__seconds = seconds
+
+    @property
+    def next_delay(self) -> float:
+        return self.__seconds

ldclient/impl/events/async_event_processor.py

@@ -202,13 +202,13 @@ class DefaultAsyncEventProcessor(AsyncEventProcessor):
     def __init__(self, config: AsyncConfig, http=None, dispatcher_class=None, diagnostic_accumulator=None):
         self._inbox = AsyncQueue(config.events_max_pending)
         self._inbox_full = False
-        self._flush_timer = AsyncRepeatingTask("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush)
-        self._contexts_flush_timer = AsyncRepeatingTask("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts)
+        self._flush_timer = AsyncRepeatingTask.at_interval("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush)
+        self._contexts_flush_timer = AsyncRepeatingTask.at_interval("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts)
         self._flush_timer.start()
         self._contexts_flush_timer.start()
         self._diagnostic_event_timer: Optional[AsyncRepeatingTask]
         if diagnostic_accumulator is not None:
-            self._diagnostic_event_timer = AsyncRepeatingTask("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic)
+            self._diagnostic_event_timer = AsyncRepeatingTask.at_interval("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic)
             self._diagnostic_event_timer.start()
         else:
             self._diagnostic_event_timer = None

ldclient/impl/events/event_processor.py

@@ -176,12 +176,12 @@ class DefaultEventProcessor(EventProcessor):
     def __init__(self, config, http=None, dispatcher_class=None, diagnostic_accumulator=None):
         self._inbox = queue.Queue(config.events_max_pending)
         self._inbox_full = False
-        self._flush_timer = RepeatingTask("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush)
-        self._contexts_flush_timer = RepeatingTask("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts)
+        self._flush_timer = RepeatingTask.at_interval("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush)
+        self._contexts_flush_timer = RepeatingTask.at_interval("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts)
         self._flush_timer.start()
         self._contexts_flush_timer.start()
         if diagnostic_accumulator is not None:
-            self._diagnostic_event_timer = RepeatingTask("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic)
+            self._diagnostic_event_timer = RepeatingTask.at_interval("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic)
             self._diagnostic_event_timer.start()
         else:
             self._diagnostic_event_timer = None

ldclient/impl/integrations/files/file_data_source.py

@@ -183,7 +183,7 @@ def __init__(self, resolved_paths, reloader, interval):
             self._paths = resolved_paths
             self._reloader = reloader
             self._file_times = self._check_file_times()
-            self._timer = RepeatingTask("ldclient.datasource.file.poll", interval, interval, self._poll)
+            self._timer = RepeatingTask.at_interval("ldclient.datasource.file.poll", interval, interval, self._poll)
             self._timer.start()
 
         def stop(self):

ldclient/impl/integrations/files/file_data_sourcev2.py

@@ -398,7 +398,7 @@ def __init__(self, resolved_paths, on_change_callback, interval):
         self._paths = resolved_paths
         self._on_change = on_change_callback
         self._file_times = self._check_file_times()
-        self._timer = RepeatingTask(
+        self._timer = RepeatingTask.at_interval(
             "ldclient.datasource.filev2.poll", interval, interval, self._poll
         )
         self._timer.start()

ldclient/impl/repeating_task.py

@@ -1,39 +1,66 @@
-import time
 from threading import Event, Thread
-from typing import Callable
+from typing import Any, Callable
 
+from ldclient.impl.delay import DelaySource, FixedDelay
 from ldclient.impl.util import log
 
 
 class RepeatingTask:
     """
-    A generic mechanism for calling a callback repeatedly at fixed intervals on a worker thread.
+    A generic mechanism for calling a callback repeatedly on a worker thread.
+
+    The wait between invocations comes from a
+    :class:`~ldclient.impl.delay.DelaySource`, which the
+    task reads after each one. Use :meth:`at_interval` for the common case of
+    a fixed interval.
     """
 
-    def __init__(self, label, interval: float, initial_delay: float, callable: Callable):
+    def __init__(self, label: str, delays: DelaySource, initial_delay: float, callable: Callable[[], Any]):
         """
         Creates the task, but does not start the worker thread yet.
 
-        :param interval: maximum time in seconds between invocations of the callback
+        :param label: names the worker thread, and appears in log messages
+        :param delays: supplies the wait after each invocation returns
         :param initial_delay: time in seconds to wait before the first invocation
-        :param callable: the function to execute repeatedly
+        :param callable: the function to execute repeatedly. Anything it
+            returns is ignored.
         """
-        self.__interval = interval
+        self.__label = label
+        self.__delays = delays
         self.__initial_delay = initial_delay
         self.__action = callable
         self.__stop = Event()
+        self.__started = False
         self.__thread = Thread(target=self._run, name=f"{label}.repeating")
         self.__thread.daemon = True
 
+    @staticmethod
+    def at_interval(label: str, interval: float, initial_delay: float, callable: Callable[[], Any]) -> 'RepeatingTask':
+        """
+        Creates a task that runs at a fixed interval.
+
+        :param interval: time in seconds to wait after each invocation returns
+        """
+        return RepeatingTask(label, FixedDelay(interval), initial_delay, callable)
+
     def start(self):
         """
-        Starts the worker thread.
+        Starts the worker thread, if it is not running already.
+
+        Starting a task twice logs and does nothing, rather than raising, so a
+        caller that is safe to call more than once stays safe.
         """
+        if self.__started:
+            log.info("Task %s has already been started; ignoring" % self.__label)
+            return
+        self.__started = True
         self.__thread.start()
 
     def stop(self):
         """
-        Tells the worker thread to stop. It cannot be restarted after this.
+        Tells the worker thread to stop.
+
+        The stop is permanent. A later :meth:`start` does not resume the task.
         """
         self.__stop.set()
 
@@ -43,10 +70,11 @@ def _run(self):
                 return
         stopped = self.__stop.is_set()
         while not stopped:
-            next_time = time.time() + self.__interval
             try:
                 self.__action()
             except Exception as e:
                 log.exception("Unexpected exception on worker thread: %s" % e)
-            delay = next_time - time.time()
+            # The wait starts when the callback returns, so a slow callback
+            # never shortens it.
+            delay = self.__delays.next_delay
             stopped = self.__stop.wait(delay) if delay > 0 else self.__stop.is_set()

ldclient/testing/impl/datasource/test_async_polling.py

@@ -426,16 +426,17 @@ async def test_initialized_returns_false_before_first_poll(self):
 
     @pytest.mark.asyncio
     @patch('ldclient.config.Config.poll_interval', new_callable=MagicMock)
-    async def test_second_start_call_raises(self, mock_interval):
+    async def test_second_start_call_is_a_no_op(self, mock_interval):
         mock_interval.__get__ = MagicMock(return_value=0)
 
         processor = make_processor()
         processor._requester.get_all_data = AsyncMock(return_value=SAMPLE_DATA)
 
         processor.start()
-        # Like a thread, the polling task can only be started once
-        with pytest.raises(RuntimeError):
-            processor.start()
+        task = processor._task
+        # The task guards against a second start; it logs and does nothing.
+        processor.start()
+        assert processor._task is task
 
         await processor.stop()
 

ldclient/testing/impl/test_delay.py

@@ -0,0 +1,7 @@
+from ldclient.impl.delay import FixedDelay
+
+
+def test_fixed_delay_always_gives_the_same_wait():
+    delays = FixedDelay(2.5)
+    assert delays.next_delay == 2.5
+    assert delays.next_delay == 2.5