From 8942e23a6941dc740f6f703f7e353f273874f104 Mon Sep 17 00:00:00 2001
From: 14mRh4X0r <14mRh4X0r@gmail.com>
Date: Mon, 7 Jun 2021 16:42:05 +0200
Subject: [PATCH] Always update AS last_pos, even on no events (#10107)

Fixes #1834.

`get_new_events_for_appservice` internally calls `get_events_as_list`, which will filter out any rejected events. If all returned events are filtered out, `_notify_interested_services` will return without updating the last handled stream position. If there are 100 consecutive such events, processing will halt altogether.

Breaking the loop is now done by checking whether we're up-to-date with `current_max` in the loop condition, instead of relying on an empty `events` list.


Signed-off-by: Willem Mulder <14mRh4X0r@gmail.com>
---
 changelog.d/10107.bugfix          |  1 +
 synapse/handlers/appservice.py    | 25 ++++++++++++-------------
 tests/handlers/test_appservice.py |  6 ++----
 3 files changed, 15 insertions(+), 17 deletions(-)
 create mode 100644 changelog.d/10107.bugfix

diff --git a/changelog.d/10107.bugfix b/changelog.d/10107.bugfix
new file mode 100644
index 000000000000..80030efab27c
--- /dev/null
+++ b/changelog.d/10107.bugfix
@@ -0,0 +1 @@
+Fixed a bug that could cause Synapse to stop notifying application services. Contributed by Willem Mulder.
diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index 177310f0beaa..862638cc4f15 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -87,7 +87,8 @@ async def _notify_interested_services(self, max_token: RoomStreamToken):
             self.is_processing = True
             try:
                 limit = 100
-                while True:
+                upper_bound = -1
+                while upper_bound < self.current_max:
                     (
                         upper_bound,
                         events,
@@ -95,9 +96,6 @@ async def _notify_interested_services(self, max_token: RoomStreamToken):
                         self.current_max, limit
                     )
 
-                    if not events:
-                        break
-
                     events_by_room = {}  # type: Dict[str, List[EventBase]]
                     for event in events:
                         events_by_room.setdefault(event.room_id, []).append(event)
@@ -153,9 +151,6 @@ async def handle_room_events(events):
 
                     await self.store.set_appservice_last_pos(upper_bound)
 
-                    now = self.clock.time_msec()
-                    ts = await self.store.get_received_ts(events[-1].event_id)
-
                     synapse.metrics.event_processing_positions.labels(
                         "appservice_sender"
                     ).set(upper_bound)
@@ -168,12 +163,16 @@ async def handle_room_events(events):
 
                     event_processing_loop_counter.labels("appservice_sender").inc()
 
-                    synapse.metrics.event_processing_lag.labels(
-                        "appservice_sender"
-                    ).set(now - ts)
-                    synapse.metrics.event_processing_last_ts.labels(
-                        "appservice_sender"
-                    ).set(ts)
+                    if events:
+                        now = self.clock.time_msec()
+                        ts = await self.store.get_received_ts(events[-1].event_id)
+
+                        synapse.metrics.event_processing_lag.labels(
+                            "appservice_sender"
+                        ).set(now - ts)
+                        synapse.metrics.event_processing_last_ts.labels(
+                            "appservice_sender"
+                        ).set(ts)
             finally:
                 self.is_processing = False
 
diff --git a/tests/handlers/test_appservice.py b/tests/handlers/test_appservice.py
index b037b12a0f3d..5d6cc2885f53 100644
--- a/tests/handlers/test_appservice.py
+++ b/tests/handlers/test_appservice.py
@@ -57,10 +57,10 @@ def test_notify_interested_services(self):
             sender="@someone:anywhere", type="m.room.message", room_id="!foo:bar"
         )
         self.mock_store.get_new_events_for_appservice.side_effect = [
-            make_awaitable((0, [event])),
             make_awaitable((0, [])),
+            make_awaitable((1, [event])),
         ]
-        self.handler.notify_interested_services(RoomStreamToken(None, 0))
+        self.handler.notify_interested_services(RoomStreamToken(None, 1))
 
         self.mock_scheduler.submit_event_for_as.assert_called_once_with(
             interested_service, event
@@ -77,7 +77,6 @@ def test_query_user_exists_unknown_user(self):
         self.mock_as_api.query_user.return_value = make_awaitable(True)
         self.mock_store.get_new_events_for_appservice.side_effect = [
             make_awaitable((0, [event])),
-            make_awaitable((0, [])),
         ]
 
         self.handler.notify_interested_services(RoomStreamToken(None, 0))
@@ -95,7 +94,6 @@ def test_query_user_exists_known_user(self):
         self.mock_as_api.query_user.return_value = make_awaitable(True)
         self.mock_store.get_new_events_for_appservice.side_effect = [
             make_awaitable((0, [event])),
-            make_awaitable((0, [])),
         ]
 
         self.handler.notify_interested_services(RoomStreamToken(None, 0))