wip

Author: jason-famedly

famedly_control_synapse/sync.py

@@ -15,6 +15,7 @@
 import logging
 
 from synapse.module_api import ModuleApi
+from synapse.util.duration import Duration
 
 from famedly_control_synapse.client import DiffRecord, FamedlyControlClient, Membership
 from famedly_control_synapse.config import FamedlyControlConfig
@@ -24,6 +25,10 @@
 logger = logging.getLogger(__name__)
 
 
+class NoTokenException(Exception):
+    """Raise when a sync token was not provided"""
+
+
 class GroupMembershipSyncer:
     """Background task that polls Famedly Control for group membership changes
     and updates managed room memberships accordingly."""
@@ -57,6 +62,7 @@ def start(self) -> None:
         """
         if self._is_running or not self.is_enabled:
             return
+        # Mark the loop as starting early, to avoid multiple background loops
         self._is_running = True
         self.api.run_as_background_process(
             "famedly_control_group_membership_sync", self.start_sync_loop
@@ -67,24 +73,43 @@ async def start_sync_loop(self) -> None:
 
         If the sync token hasn't been loaded yet, tries to read it from the
         database.  If it's still not available (no managed room created yet),
-        returns and will be retried on the next api request. Otherwise sets
+        returns and will be retried on the next api request. Otherwise, sets
         up the periodic ``_process_sync``.
         """
         entry = await self.repository.get_sync_token_entry()
         if entry is None:
             self._is_running = False
             return
+
+        logger.info("Starting loop to poll /get_all_groups_diffs")
+
         self._sync_token_user_id, self._sync_token = entry
-        self.api.looping_background_call(
-            self._process_sync,
-            self.polling_interval_seconds * 1000,
-            desc="famedly_control_group_membership_sync",
-        )
+
+        while self._is_running:
+            try:
+                await self._process_sync()
+            # Several kinds of exceptions can be raised here. Timeouts, Cancellations,
+            # Network Exceptions, HTTP Exceptions, general exceptions from database or
+            # IO, etc. I believe the CancelledError from twisted.defer can be raised
+            # from the sleep() below. Perhaps in the future that should be caught, it
+            # should only occur during server shutdown though and can safely ignored
+            # (although it may look rather scary in the logs)
+            except Exception as e:
+                logger.error("Exception during loop: %r", e)
+                await self.api._hs.get_clock().sleep(
+                    Duration(seconds=self.polling_interval_seconds)
+                )
 
     async def _process_sync(self) -> None:
-        """Execute a single sync iteration: fetch diffs and apply membership changes."""
+        """
+        Execute a single sync iteration: fetch diffs and apply membership changes.
+
+        Long poll for the response. If a 'next_sync' token is returned, use that on the
+        next request.
+        If no token is returned, wait for the configured amount and try again.
+        """
         response = await self.client.get_all_groups_diffs(
-            sync=self._sync_token, timeout=30
+            sync=self._sync_token, timeout=self.polling_interval_seconds
         )
 
         sync_succeeded = True
@@ -111,8 +136,15 @@ async def _process_sync(self) -> None:
             )
             return
 
+        if not response.next_sync:
+            # If there was not a 'next_sync' token, long polling is not used. Raise, and
+            # the loop can handle the wait
+            raise NoTokenException("No token in response")
+
         self._sync_token = response.next_sync
-        if self._sync_token_user_id is not None:
+        if self._sync_token_user_id is None:
+            logger.error("Loop was running but there was no user id for the sync token")
+        else:
             await self.repository.set_sync_token(
                 self._sync_token_user_id, self._sync_token
             )