Port call_later

Author: erikjohnston

synapse/app/phone_stats_home.py

@@ -278,14 +278,14 @@ async def _generate_monthly_active_users() -> None:
         # We need to defer this init for the cases that we daemonize
         # otherwise the process ID we get is that of the non-daemon process
         clock.call_later(
-            0,
+            Duration(seconds=0),
             performance_stats_init,
         )
 
         # We wait 5 minutes to send the first set of stats as the server can
         # be quite busy the first few minutes
         clock.call_later(
-            INITIAL_DELAY_BEFORE_FIRST_PHONE_HOME.as_secs(),
+            INITIAL_DELAY_BEFORE_FIRST_PHONE_HOME,
             phone_stats_home,
             hs,
             stats,

synapse/handlers/delayed_events.py

@@ -42,6 +42,7 @@
     UserID,
     create_requester,
 )
+from synapse.util.duration import Duration
 from synapse.util.events import generate_fake_event_id
 from synapse.util.metrics import Measure
 from synapse.util.sentinel import Sentinel
@@ -92,7 +93,7 @@ async def _schedule_db_events() -> None:
                 # Kick off again (without blocking) to catch any missed notifications
                 # that may have fired before the callback was added.
                 self._clock.call_later(
-                    0,
+                    Duration(seconds=0),
                     self.notify_new_event,
                 )
 
@@ -501,17 +502,17 @@ def _schedule_next_at_or_none(self, next_send_ts: Timestamp | None) -> None:
 
     def _schedule_next_at(self, next_send_ts: Timestamp) -> None:
         delay = next_send_ts - self._get_current_ts()
-        delay_sec = delay / 1000 if delay > 0 else 0
+        delay_duration = Duration(milliseconds=max(delay, 0))
 
         if self._next_delayed_event_call is None:
             self._next_delayed_event_call = self._clock.call_later(
-                delay_sec,
+                delay_duration,
                 self.hs.run_as_background_process,
                 "_send_on_timeout",
                 self._send_on_timeout,
             )
         else:
-            self._next_delayed_event_call.reset(delay_sec)
+            self._next_delayed_event_call.reset(delay_duration.as_secs())
 
     async def get_all_for_user(self, requester: Requester) -> list[JsonDict]:
         """Return all pending delayed events requested by the given user."""

synapse/handlers/message.py

@@ -434,14 +434,11 @@ def _schedule_expiry_for_event(self, event_id: str, expiry_ts: int) -> None:
 
         # Figure out how many seconds we need to wait before expiring the event.
         now_ms = self.clock.time_msec()
-        delay = (expiry_ts - now_ms) / 1000
+        delay = Duration(milliseconds=max(expiry_ts - now_ms, 0))
 
-        # callLater doesn't support negative delays, so trim the delay to 0 if we're
-        # in that case.
-        if delay < 0:
-            delay = 0
-
-        logger.info("Scheduling expiry for event %s in %.3fs", event_id, delay)
+        logger.info(
+            "Scheduling expiry for event %s in %.3fs", event_id, delay.as_secs()
+        )
 
         self._scheduled_expiry = self.clock.call_later(
             delay,

synapse/handlers/presence.py

@@ -862,7 +862,7 @@ def __init__(self, hs: "HomeServer"):
             # The initial delay is to allow disconnected clients a chance to
             # reconnect before we treat them as offline.
             self.clock.call_later(
-                30,
+                Duration(seconds=30),
                 self.clock.looping_call,
                 self._handle_timeouts,
                 Duration(seconds=5),
@@ -872,7 +872,7 @@ def __init__(self, hs: "HomeServer"):
         # internally.
         if self._presence_enabled:
             self.clock.call_later(
-                60,
+                Duration(seconds=60),
                 self.clock.looping_call,
                 self._persist_unpersisted_changes,
                 Duration(minutes=1),

synapse/handlers/room_member.py

@@ -2191,7 +2191,7 @@ def __init__(self, hs: "HomeServer"):
 
             # We kick this off to pick up outstanding work from before the last restart.
             self._clock.call_later(
-                0,
+                Duration(seconds=0),
                 self.notify_new_event,
             )
 

synapse/handlers/stats.py

@@ -32,6 +32,7 @@
 from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions
 from synapse.storage.databases.main.state_deltas import StateDelta
 from synapse.types import JsonDict
+from synapse.util.duration import Duration
 from synapse.util.events import get_plain_text_topic_from_event_content
 
 if TYPE_CHECKING:
@@ -72,7 +73,7 @@ def __init__(self, hs: "HomeServer"):
             # We kick this off so that we don't have to wait for a change before
             # we start populating stats
             self.clock.call_later(
-                0,
+                Duration(seconds=0),
                 self.notify_new_event,
             )
 

synapse/handlers/user_directory.py

@@ -40,6 +40,7 @@
 from synapse.storage.databases.main.user_directory import SearchResult
 from synapse.storage.roommember import ProfileInfo
 from synapse.types import UserID
+from synapse.util.duration import Duration
 from synapse.util.metrics import Measure
 from synapse.util.retryutils import NotRetryingDestination
 from synapse.util.stringutils import non_null_str_or_none
@@ -52,15 +53,15 @@
 # Don't refresh a stale user directory entry, using a Federation /profile request,
 # for 60 seconds. This gives time for other state events to arrive (which will
 # then be coalesced such that only one /profile request is made).
-USER_DIRECTORY_STALE_REFRESH_TIME_MS = 60 * 1000
+USER_DIRECTORY_STALE_REFRESH_TIME = Duration(seconds=60)
 
 # Maximum number of remote servers that we will attempt to refresh profiles for
 # in one go.
 MAX_SERVERS_TO_REFRESH_PROFILES_FOR_IN_ONE_GO = 5
 
 # As long as we have servers to refresh (without backoff), keep adding more
 # every 15 seconds.
-INTERVAL_TO_ADD_MORE_SERVERS_TO_REFRESH_PROFILES = 15
+INTERVAL_TO_ADD_MORE_SERVERS_TO_REFRESH_PROFILES = Duration(seconds=15)
 
 
 def calculate_time_of_next_retry(now_ts: int, retry_count: int) -> int:
@@ -137,13 +138,13 @@ def __init__(self, hs: "HomeServer"):
             # We kick this off so that we don't have to wait for a change before
             # we start populating the user directory
             self.clock.call_later(
-                0,
+                Duration(seconds=0),
                 self.notify_new_event,
             )
 
             # Kick off the profile refresh process on startup
             self._refresh_remote_profiles_call_later = self.clock.call_later(
-                10,
+                Duration(seconds=10),
                 self.kick_off_remote_profile_refresh_process,
             )
 
@@ -550,21 +551,21 @@ async def _handle_possible_remote_profile_change(
             now_ts = self.clock.time_msec()
             await self.store.set_remote_user_profile_in_user_dir_stale(
                 user_id,
-                next_try_at_ms=now_ts + USER_DIRECTORY_STALE_REFRESH_TIME_MS,
+                next_try_at_ms=now_ts + USER_DIRECTORY_STALE_REFRESH_TIME.as_millis(),
                 retry_counter=0,
             )
             # Schedule a wake-up to refresh the user directory for this server.
             # We intentionally wake up this server directly because we don't want
             # other servers ahead of it in the queue to get in the way of updating
             # the profile if the server only just sent us an event.
             self.clock.call_later(
-                USER_DIRECTORY_STALE_REFRESH_TIME_MS // 1000 + 1,
+                USER_DIRECTORY_STALE_REFRESH_TIME + Duration(seconds=1),
                 self.kick_off_remote_profile_refresh_process_for_remote_server,
                 UserID.from_string(user_id).domain,
             )
             # Schedule a wake-up to handle any backoffs that may occur in the future.
             self.clock.call_later(
-                2 * USER_DIRECTORY_STALE_REFRESH_TIME_MS // 1000 + 1,
+                USER_DIRECTORY_STALE_REFRESH_TIME * 2 + Duration(seconds=1),
                 self.kick_off_remote_profile_refresh_process,
             )
             return
@@ -656,7 +657,9 @@ async def _unsafe_refresh_remote_profiles(self) -> None:
                 if not users:
                     return
                 _, _, next_try_at_ts = users[0]
-                delay = ((next_try_at_ts - self.clock.time_msec()) // 1000) + 2
+                delay = Duration(
+                    milliseconds=next_try_at_ts - self.clock.time_msec()
+                ) + Duration(seconds=2)
                 self._refresh_remote_profiles_call_later = self.clock.call_later(
                     delay,
                     self.kick_off_remote_profile_refresh_process,

synapse/handlers/worker_lock.py

@@ -187,7 +187,7 @@ def _wake_all_locks(
                 lock.release_lock()
 
         self._clock.call_later(
-            0,
+            Duration(seconds=0),
             _wake_all_locks,
             locks,
         )

synapse/http/client.py

@@ -87,6 +87,7 @@
 from synapse.types import ISynapseReactor, StrSequence
 from synapse.util.async_helpers import timeout_deferred
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 from synapse.util.json import json_decoder
 
 if TYPE_CHECKING:
@@ -172,7 +173,7 @@ def _make_scheduler(clock: Clock) -> Callable[[Callable[[], object]], IDelayedCa
 
     def _scheduler(x: Callable[[], object]) -> IDelayedCall:
         return clock.call_later(
-            _EPSILON,
+            Duration(seconds=_EPSILON),
             x,
         )
 

synapse/module_api/__init__.py

@@ -1445,8 +1445,7 @@ def delayed_background_call(
             desc = f.__name__
 
         return self._clock.call_later(
-            # convert ms to seconds as needed by call_later.
-            msec * 0.001,
+            Duration(milliseconds=msec),
             self._hs.run_as_background_process,
             desc,
             lambda: maybe_awaitable(f(*args, **kwargs)),