test: Pin gap semantics and delay-source read ordering

Author: jsonbailey

ldclient/impl/aio/concurrency.py

@@ -215,6 +215,7 @@ def start(self):
             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:

ldclient/testing/impl/test_repeating_task.py

@@ -79,35 +79,70 @@ def test_task_executes_until_stopped():
     assert no_more_items is True
 
 
-class _MutableDelay(DelaySource):
-    """A delay source a test can move between invocations."""
+class _RecordingDelay(DelaySource):
+    """A delay source that records each read, so a test can see when the task
+    asks for a wait rather than only what the action saw."""
 
-    def __init__(self, seconds: float):
+    def __init__(self, seconds: float, events: list):
         self.seconds = seconds
+        self._events = events
 
     @property
     def next_delay(self) -> float:
+        self._events.append(('read', self.seconds))
         return self.seconds
 
 
 def test_the_task_reads_the_delay_source_after_every_invocation():
-    """A value the action decides takes effect on the next wait."""
-    reads = Queue()
-    delays = _MutableDelay(0.01)
+    """One read per invocation, after it. A task that read the source once up
+    front would show a read before the first invocation, and would never see
+    the value the action set."""
+    events: list = []
+    delays = _RecordingDelay(0.01, events)
 
     def do_task():
-        reads.put(delays.seconds)
-        delays.seconds = 0.02  # what the next wait must use
+        events.append('invoke')
+        delays.seconds = 0.02
 
-    task = RepeatingTask("ldclient.testing.mutable-delay", delays, 0, do_task)
+    task = RepeatingTask("ldclient.testing.recording-delay", delays, 0, do_task)
     try:
         task.start()
-        assert reads.get(True, 1) == 0.01
-        assert reads.get(True, 1) == 0.02
-        assert reads.get(True, 1) == 0.02
+        deadline = time.time() + 2
+        while events.count('invoke') < 3 and time.time() < deadline:
+            time.sleep(0.005)
     finally:
         task.stop()
 
+    # Reads and invocations alternate, starting with an invocation, and every
+    # read sees 0.02 -- the initial 0.01 is never read.
+    assert events[:5] == ['invoke', ('read', 0.02), 'invoke', ('read', 0.02), 'invoke']
+
+
+def test_the_interval_starts_when_the_callback_returns():
+    """A slow callback must not shorten its own wait: one invocation to the
+    next is the interval plus however long the callback took."""
+    work = 0.15
+    interval = 0.15
+    starts = Queue()
+
+    def do_task():
+        starts.put(time.time())
+        time.sleep(work)
+
+    task = RepeatingTask.at_interval("ldclient.testing.slow-callback", interval, 0, do_task)
+    try:
+        first = None
+        task.start()
+        first = starts.get(True, 2)
+        second = starts.get(True, 2)
+    finally:
+        task.stop()
+
+    # Measuring the interval from the start of the callback would give about
+    # `interval`; measuring from its return gives interval + work. The 10%
+    # slack is for scheduling noise, and leaves the two regimes far apart.
+    assert (second - first) >= (interval + work) * 0.9
+
 
 def test_whatever_the_action_returns_is_ignored():
     """Guards big-segment polling, whose action returns a status object."""

ldclient/testing/test_aio.py

@@ -177,6 +177,48 @@ async def action():
         await asyncio.sleep(0.1)
         assert counts['n'] == 1
 
+    @pytest.mark.asyncio
+    async def test_async_a_task_that_dies_is_logged(self, caplog):
+        """Without a done callback the held reference suppresses asyncio's own
+        warning, so a dead loop would be entirely silent."""
+        caplog.set_level(logging.ERROR)
+
+        class Exploding:
+            @property
+            def next_delay(self):
+                raise RuntimeError("delay source is broken")
+
+        async def action():
+            pass
+
+        task = aio.AsyncRepeatingTask("test.repeating", Exploding(), 0, action)
+        task.start()
+        await _async_wait_until(lambda: caplog.records, timeout=2)
+        task.stop()
+
+        assert "Unhandled exception in background task" in caplog.records[0].getMessage()
+
+    @pytest.mark.asyncio
+    async def test_async_interval_starts_when_the_callback_returns(self):
+        """A slow callback must not shorten its own wait: one invocation to the
+        next is the interval plus however long the callback took."""
+        work = 0.15
+        interval = 0.15
+        starts: list = []
+
+        async def action():
+            starts.append(time.time())
+            await asyncio.sleep(work)
+
+        task = aio.AsyncRepeatingTask.at_interval("test.repeating", interval, 0, action)
+        task.start()
+        await _async_wait_until(lambda: len(starts) >= 2, timeout=3)
+        task.stop()
+
+        # Measuring the interval from the start of the callback would give
+        # about `interval`; measuring from its return gives interval + work.
+        assert (starts[1] - starts[0]) >= (interval + work) * 0.9
+
     @pytest.mark.asyncio
     async def test_async_second_start_logs_and_does_not_raise(self, caplog):
         """Mirrors the sync primitive. A raise here can surface out of a caller