@@ -1,7 +1,6 @@
 import functools
 import json
 import logging
-import time
 from urllib.parse import urljoin
 
 import httpx
@@ -10,6 +9,7 @@
 from ably.http.httputils import HttpUtils
 from ably.rest.auth import Auth
 from ably.transport.defaults import Defaults
+from ably.util.clock import select_clock
 from ably.util.exceptions import AblyException
 from ably.util.helper import extract_url_params, is_token_error
 
@@ -129,6 +129,7 @@ def __init__(self, ably, options):
         # Cached fallback host (RSC15f)
         self.__host = None
         self.__host_expires = None
+        self.__clock = select_clock(options)
         self.__client = self.__create_client(options)
 
     @staticmethod
@@ -154,7 +155,7 @@ def get_hosts(self):
             return hosts
 
         # unstore saved fallback host after fallbackRetryTimeout (RSC15f)
-        if self.__host_expires is not None and time.time() > self.__host_expires:
+        if self.__host_expires is not None and self.__clock.now_ms() > self.__host_expires:
             self.__host = None
             self.__host_expires = None
             return hosts
@@ -191,12 +192,12 @@ async def make_request(self, method, path, version=None, headers=None, body=None
 
         timeout = (self.http_open_timeout, self.http_request_timeout)
         http_max_retry_duration = self.http_max_retry_duration
-        requested_at = time.time()
+        requested_at = self.__clock.now_ms()
 
         hosts = self.get_hosts()
         for retry_count, host in enumerate(hosts):
             def should_stop_retrying(retry_count=retry_count):
-                time_passed = time.time() - requested_at
+                time_passed = (self.__clock.now_ms() - requested_at) / 1000
                 # if it's the last try or cumulative timeout is done, we stop retrying
                 return retry_count == len(hosts) - 1 or time_passed > http_max_retry_duration
 
@@ -238,7 +239,7 @@ def should_stop_retrying(retry_count=retry_count):
                     # Keep fallback host for later (RSC15f)
                     if retry_count > 0 and host != self.options.get_host():
                         self.__host = host
-                        self.__host_expires = time.time() + (self.options.fallback_retry_timeout / 1000.0)
+                        self.__host_expires = self.__clock.now_ms() + self.options.fallback_retry_timeout
 
                     return Response(response)
                 except AblyException as e:
 
@@ -2,7 +2,6 @@
 
 import base64
 import logging
-import time
 import uuid
 from datetime import timedelta
 from typing import TYPE_CHECKING
@@ -18,6 +17,7 @@
 from ably.types.capability import Capability
 from ably.types.tokendetails import TokenDetails
 from ably.types.tokenrequest import TokenRequest
+from ably.util.clock import select_clock
 from ably.util.exceptions import AblyAuthException, AblyException, IncompatibleClientIdException
 from ably.util.helper import extract_url_params
 
@@ -34,6 +34,7 @@ class Method:
     def __init__(self, ably: AblyRest | AblyRealtime, options: Options):
         self.__ably = ably
         self.__auth_options = options
+        self.__clock = select_clock(options)
 
         if not self.ably._is_realtime:
             self.__client_id = options.client_id
@@ -383,7 +384,7 @@ async def _get_auth_headers(self):
 
     def _timestamp(self):
         """Returns the local time in milliseconds since the unix epoch"""
-        return int(time.time() * 1000)
+        return self.__clock.now_ms()
 
     def _random_nonce(self):
         return uuid.uuid4().hex[:16]
 
@@ -6,11 +6,17 @@ class TestOptions:
     :Parameters:
       - `http_transport`: an `httpx.AsyncBaseTransport` which handles every
         HTTP request the client makes, in place of the network.
+      - `clock`: the source of time the client reads, in place of
+        `ably.util.clock.Clock`. It supplies `now_ms()`, `monotonic_ms()` and
+        `timer(timeout_ms, callback)`, where `callback` is either a coroutine
+        function or a plain callable and the return value has a `cancel()`
+        method.
     """
 
     # Excludes the class from pytest collection, which would otherwise treat
     # any module importing it as declaring a test suite.
     __test__ = False
 
-    def __init__(self, http_transport=None):
+    def __init__(self, http_transport=None, clock=None):
         self.http_transport = http_transport
+        self.clock = clock
@@ -0,0 +1,41 @@
+import time
+from typing import Callable
+
+from ably.util.helper import Timer
+
+
+class Clock:
+    """The source of time for every decision a client makes from it.
+
+    Token expiry, the server-time offset and the fallback-host cache all read
+    the clock, and delayed callbacks are scheduled through it, so replacing one
+    moves all of them together.
+
+    `now_ms` is the time of day and may step; `monotonic_ms` only ever moves
+    forward and is what a duration is measured with.
+    """
+
+    def now_ms(self) -> int:
+        """The milliseconds elapsed since the unix epoch."""
+        return round(time.time_ns() / 1_000_000)
+
+    def monotonic_ms(self) -> float:
+        """The milliseconds elapsed since an arbitrary fixed point."""
+        return time.monotonic() * 1000
+
+    def timer(self, timeout: float, callback: Callable) -> Timer:
+        """Schedules `callback` for `timeout` milliseconds from now."""
+        return Timer(timeout, callback)
+
+
+def select_clock(options) -> Clock:
+    """The clock a client reads its time from.
+
+    `TestOptions.clock` substitutes for the real one during tests, letting them
+    drive time-dependent behaviour without waiting for it. Clients which supply
+    none get `Clock`.
+    """
+    test_options = getattr(options, 'test_options', None)
+    if test_options is not None and test_options.clock is not None:
+        return test_options.clock
+    return Clock()
@@ -0,0 +1,124 @@
+import httpx
+import pytest
+
+from ably import AblyRest
+from ably.types.testoptions import TestOptions
+from ably.util.clock import Clock
+from ably.util.exceptions import AblyException
+
+# A plausible wall-clock reading for a fake to start from, so that a time taken
+# against it looks like a real one.
+EPOCH_MS = 1_700_000_000_000
+
+
+class FakeClock:
+    """A clock whose notional time only the test moves.
+
+    `advance` moves it by hand; `step_ms` moves it by that much on every read,
+    which is how a duration measured across several reads is driven.
+    """
+
+    def __init__(self, now_ms=EPOCH_MS, step_ms=0):
+        self.__now_ms = now_ms
+        self.__step_ms = step_ms
+
+    def now_ms(self):
+        reading = self.__now_ms
+        self.__now_ms += self.__step_ms
+        return reading
+
+    def monotonic_ms(self):
+        return float(self.__now_ms)
+
+    def advance(self, milliseconds):
+        self.__now_ms += milliseconds
+
+    def timer(self, timeout, callback):
+        raise AssertionError('the REST client schedules no delayed callbacks')
+
+
+class RecordingTransport(httpx.AsyncBaseTransport):
+    """Records the host of every request and refuses the ones in `refusing`.
+
+    A host left out of `refusing` is served a server time. `refusing` is set
+    after the client is built, since the host to refuse is one of its own.
+    """
+
+    def __init__(self):
+        self.refusing = ()
+        self.hosts = []
+
+    async def handle_async_request(self, request):
+        self.hosts.append(request.url.host)
+        if request.url.host in self.refusing:
+            raise httpx.ConnectError('connection refused', request=request)
+        return httpx.Response(200, json=[1500000000000])
+
+
+def test_auth_timestamps_from_the_injected_clock():
+    ably = AblyRest(key='name:secret', test_options=TestOptions(clock=FakeClock(1_500_000_000_000)))
+    assert ably.auth._timestamp() == 1_500_000_000_000
+
+
+def test_a_client_without_test_options_reads_the_real_clock():
+    ably = AblyRest(token='foo')
+    assert isinstance(ably.auth._Auth__clock, Clock)
+    assert isinstance(ably.http._Http__clock, Clock)
+
+
+async def test_the_cached_fallback_host_expires_on_the_clock():
+    # RSC15f: the cache lasts fallback_retry_timeout milliseconds
+    clock = FakeClock()
+    transport = RecordingTransport()
+    ably = AblyRest(token='foo', fallback_retry_timeout=2000,
+                    test_options=TestOptions(http_transport=transport, clock=clock))
+    primary = ably.options.get_host()
+    transport.refusing = (primary,)
+
+    await ably.time()
+
+    fallback = transport.hosts[-1]
+    assert fallback != primary
+    assert ably.http._Http__host_expires == EPOCH_MS + 2000
+    assert ably.http.get_hosts()[0] == fallback
+
+    clock.advance(1999)
+    assert ably.http.get_hosts()[0] == fallback
+
+    clock.advance(2)
+    assert ably.http.get_hosts()[0] == primary
+    assert ably.http._Http__host_expires is None
+
+    await ably.close()
+
+
+async def test_every_host_is_tried_while_the_retry_budget_lasts():
+    # A second of clock time per reading, against the default 15 second budget
+    clock = FakeClock(step_ms=1000)
+    transport = RecordingTransport()
+    ably = AblyRest(token='foo', test_options=TestOptions(http_transport=transport, clock=clock))
+    hosts = ably.http.get_hosts()
+    transport.refusing = tuple(hosts)
+
+    with pytest.raises(AblyException):
+        await ably.time()
+
+    assert transport.hosts == hosts
+
+    await ably.close()
+
+
+async def test_retrying_stops_once_the_retry_budget_is_spent():
+    # RSC15l: http_max_retry_duration is seconds where the clock is milliseconds
+    clock = FakeClock(step_ms=1000)
+    transport = RecordingTransport()
+    ably = AblyRest(token='foo', http_max_retry_duration=0.5,
+                    test_options=TestOptions(http_transport=transport, clock=clock))
+    transport.refusing = tuple(ably.http.get_hosts())
+
+    with pytest.raises(AblyException):
+        await ably.time()
+
+    assert len(transport.hosts) == 1
+
+    await ably.close()