diff --git a/ldclient/impl/aio/concurrency.py b/ldclient/impl/aio/concurrency.py
index ed3f7962..ab1f6220 100644
--- a/ldclient/impl/aio/concurrency.py
+++ b/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:
diff --git a/ldclient/impl/async_big_segments.py b/ldclient/impl/async_big_segments.py
index ce4aecd0..16315e0a 100644
--- a/ldclient/impl/async_big_segments.py
+++ b/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
diff --git a/ldclient/impl/big_segments.py b/ldclient/impl/big_segments.py
index cf2dec61..2d5dcf43 100644
--- a/ldclient/impl/big_segments.py
+++ b/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):
diff --git a/ldclient/impl/datasource/async_polling.py b/ldclient/impl/datasource/async_polling.py
index d0530a7c..2c2e56e7 100644
--- a/ldclient/impl/datasource/async_polling.py
+++ b/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))
diff --git a/ldclient/impl/datasource/polling.py b/ldclient/impl/datasource/polling.py
index 171df9eb..8b33bc3d 100644
--- a/ldclient/impl/datasource/polling.py
+++ b/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))
diff --git a/ldclient/impl/datasystem/async_fdv2.py b/ldclient/impl/datasystem/async_fdv2.py
index c9d0388d..612fe4cb 100644
--- a/ldclient/impl/datasystem/async_fdv2.py
+++ b/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,
diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py
index 38655415..09ca6939 100644
--- a/ldclient/impl/datasystem/fdv1.py
+++ b/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
diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py
index 3455eab1..75f8d28c 100644
--- a/ldclient/impl/datasystem/fdv2.py
+++ b/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,
diff --git a/ldclient/impl/delay.py b/ldclient/impl/delay.py
new file mode 100644
index 00000000..b0ebb395
--- /dev/null
+++ b/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
diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py
index 1e5b0215..dafd2483 100644
--- a/ldclient/impl/events/async_event_processor.py
+++ b/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
diff --git a/ldclient/impl/events/event_processor.py b/ldclient/impl/events/event_processor.py
index 20cf03a4..6581070a 100644
--- a/ldclient/impl/events/event_processor.py
+++ b/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
diff --git a/ldclient/impl/integrations/files/file_data_source.py b/ldclient/impl/integrations/files/file_data_source.py
index 0fd0593c..81fb38d3 100644
--- a/ldclient/impl/integrations/files/file_data_source.py
+++ b/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):
diff --git a/ldclient/impl/integrations/files/file_data_sourcev2.py b/ldclient/impl/integrations/files/file_data_sourcev2.py
index 5442b81e..032fc3fb 100644
--- a/ldclient/impl/integrations/files/file_data_sourcev2.py
+++ b/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()
diff --git a/ldclient/impl/repeating_task.py b/ldclient/impl/repeating_task.py
index 2d65de87..d2e4abe1 100644
--- a/ldclient/impl/repeating_task.py
+++ b/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()
diff --git a/ldclient/testing/impl/datasource/test_async_polling.py b/ldclient/testing/impl/datasource/test_async_polling.py
index c639efb2..919bf944 100644
--- a/ldclient/testing/impl/datasource/test_async_polling.py
+++ b/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()
 
diff --git a/ldclient/testing/impl/test_delay.py b/ldclient/testing/impl/test_delay.py
new file mode 100644
index 00000000..f6684809
--- /dev/null
+++ b/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
diff --git a/ldclient/testing/impl/test_repeating_task.py b/ldclient/testing/impl/test_repeating_task.py
index 7d29cbf3..979652c4 100644
--- a/ldclient/testing/impl/test_repeating_task.py
+++ b/ldclient/testing/impl/test_repeating_task.py
@@ -1,13 +1,15 @@
+import logging
 import time
 from queue import Empty, Queue
 from threading import Event
 
+from ldclient.impl.delay import DelaySource
 from ldclient.impl.repeating_task import RepeatingTask
 
 
 def test_task_does_not_start_when_created():
     signal = Event()
-    task = RepeatingTask("ldclient.testing.set-signal", 0.01, 0, lambda: signal.set())
+    task = RepeatingTask.at_interval("ldclient.testing.set-signal", 0.01, 0, lambda: signal.set())
     try:
         signal_was_set = signal.wait(0.1)
         assert signal_was_set is False
@@ -15,9 +17,47 @@ def test_task_does_not_start_when_created():
         task.stop()
 
 
+def test_a_second_start_logs_and_does_not_raise(caplog):
+    """A raise here can surface out of a caller that is documented as safe to
+    call more than once, such as AsyncLDClient.start()."""
+    caplog.set_level(logging.INFO)
+    queue = Queue()
+    task = RepeatingTask.at_interval("ldclient.testing.enqueue-time", 0.01, 0, lambda: queue.put(time.time()))
+    try:
+        task.start()
+        thread = task._RepeatingTask__thread
+
+        task.start()
+
+        assert task._RepeatingTask__thread is thread
+        assert queue.get(True, 1) is not None  # still running
+    finally:
+        task.stop()
+
+    assert any(
+        r.getMessage() == "Task ldclient.testing.enqueue-time has already been started; ignoring"
+        for r in caplog.records
+    )
+
+
+def test_a_start_after_stop_does_not_resume_the_task():
+    counter = 0
+
+    def do_task():
+        nonlocal counter
+        counter += 1
+
+    task = RepeatingTask.at_interval("ldclient.testing.task-runner", 0.01, 0, do_task)
+    task.stop()
+    task.start()
+    time.sleep(0.1)
+
+    assert counter == 0
+
+
 def test_task_executes_until_stopped():
     queue = Queue()
-    task = RepeatingTask("ldclient.testing.enqueue-time", 0.1, 0, lambda: queue.put(time.time()))
+    task = RepeatingTask.at_interval("ldclient.testing.enqueue-time", 0.1, 0, lambda: queue.put(time.time()))
     try:
         last = None
         task.start()
@@ -39,6 +79,88 @@ def test_task_executes_until_stopped():
     assert no_more_items is True
 
 
+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, 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():
+    """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():
+        events.append('invoke')
+        delays.seconds = 0.02
+
+    task = RepeatingTask("ldclient.testing.recording-delay", delays, 0, do_task)
+    try:
+        task.start()
+        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."""
+    calls = Queue()
+
+    def do_task():
+        calls.put(time.time())
+        return object()  # not a number, and not for the task to interpret
+
+    task = RepeatingTask.at_interval("ldclient.testing.returns-a-value", 0.01, 0, do_task)
+    try:
+        task.start()
+        for _ in range(3):
+            assert calls.get(True, 1) is not None
+    finally:
+        task.stop()
+
+
 def test_task_can_be_stopped_from_within_the_task():
     counter = 0
     stopped = Event()
@@ -51,7 +173,7 @@ def do_task():
             task.stop()
             stopped.set()
 
-    task = RepeatingTask("ldclient.testing.task-runner", 0.01, 0, do_task)
+    task = RepeatingTask.at_interval("ldclient.testing.task-runner", 0.01, 0, do_task)
     try:
         task.start()
         assert stopped.wait(0.1) is True
diff --git a/ldclient/testing/test_aio.py b/ldclient/testing/test_aio.py
index 85174285..4bc2668d 100644
--- a/ldclient/testing/test_aio.py
+++ b/ldclient/testing/test_aio.py
@@ -6,6 +6,7 @@
 """
 
 import asyncio
+import logging
 import subprocess
 import sys
 import threading
@@ -127,7 +128,7 @@ async def test_async_fires_repeatedly_then_stops(self):
         async def action():
             counts['n'] += 1
 
-        task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action)
+        task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action)
         task.start()
         await _async_wait_until(lambda: counts['n'] >= 3)
         task.stop()
@@ -143,7 +144,7 @@ async def test_async_initial_delay_respected(self):
         async def action():
             counts['n'] += 1
 
-        task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0.1, action)
+        task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0.1, action)
         task.start()
         await asyncio.sleep(0.03)
         assert counts['n'] == 0
@@ -157,7 +158,7 @@ async def action():
             counts['n'] += 1
             raise RuntimeError("boom")
 
-        task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action)
+        task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action)
         task.start()
         await _async_wait_until(lambda: counts['n'] >= 2)
         task.stop()
@@ -171,22 +172,92 @@ async def action():
             counts['n'] += 1
             holder['task'].stop()
 
-        holder['task'] = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action)
+        holder['task'] = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action)
         holder['task'].start()
         await asyncio.sleep(0.1)
         assert counts['n'] == 1
 
     @pytest.mark.asyncio
-    async def test_async_second_start_raises(self):
+    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", 0.01, 0, action)
+        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()
-        with pytest.raises(RuntimeError):
-            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
+        that is documented as safe to call more than once."""
+        caplog.set_level(logging.INFO)
+        counts = {'n': 0}
+
+        async def action():
+            counts['n'] += 1
+
+        task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action)
+        task.start()
+        handle = task._AsyncRepeatingTask__task
+
+        task.start()
+
+        assert task._AsyncRepeatingTask__task is handle
+        await _async_wait_until(lambda: counts['n'] >= 1)
+        task.stop()
+
+        assert any(
+            r.getMessage() == "Task test.repeating has already been started; ignoring"
+            for r in caplog.records
+        )
+
+    @pytest.mark.asyncio
+    async def test_async_start_after_stop_does_not_resume_the_task(self):
+        counts = {'n': 0}
+
+        async def action():
+            counts['n'] += 1
+
+        task = aio.AsyncRepeatingTask.at_interval("test.repeating", 0.01, 0, action)
+        task.stop()
+        task.start()
+        await asyncio.sleep(0.05)
+
+        assert counts['n'] == 0
+
 
 class TestBoundedTaskSet:
     @pytest.mark.asyncio