From 723ce73d0253adddfb0264ff50ca4ebce0b70130 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 4 Jul 2022 16:02:21 +0100
Subject: [PATCH] Fix stuck notification counts on small servers (#13168)

---
 changelog.d/13168.bugfix                             |  1 +
 synapse/storage/databases/main/event_push_actions.py |  9 +++++++--
 tests/storage/test_event_push_actions.py             | 10 +++++-----
 3 files changed, 13 insertions(+), 7 deletions(-)
 create mode 100644 changelog.d/13168.bugfix

diff --git a/changelog.d/13168.bugfix b/changelog.d/13168.bugfix
new file mode 100644
index 000000000..f462260c5
--- /dev/null
+++ b/changelog.d/13168.bugfix
@@ -0,0 +1 @@
+Fix unread counts for users on small servers. Introduced in v1.62.0rc1.
diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py
index 7d4754b3d..505616e21 100644
--- a/synapse/storage/databases/main/event_push_actions.py
+++ b/synapse/storage/databases/main/event_push_actions.py
@@ -972,7 +972,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, EventsWorkerStore, SQLBas
         stream_row = txn.fetchone()
         if stream_row:
             (offset_stream_ordering,) = stream_row
-            rotate_to_stream_ordering = offset_stream_ordering
+
+            # We need to bound by the current token to ensure that we handle
+            # out-of-order writes correctly.
+            rotate_to_stream_ordering = min(
+                offset_stream_ordering, self._stream_id_gen.get_current_token()
+            )
             caught_up = False
         else:
             rotate_to_stream_ordering = self._stream_id_gen.get_current_token()
@@ -1004,7 +1009,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, EventsWorkerStore, SQLBas
                 SELECT user_id, room_id, count(*) as cnt,
                     max(stream_ordering) as stream_ordering
                 FROM event_push_actions
-                WHERE ? <= stream_ordering AND stream_ordering < ?
+                WHERE ? < stream_ordering AND stream_ordering <= ?
                     AND %s = 1
                 GROUP BY user_id, room_id
             ) AS upd
diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py
index 684485ae0..852b66338 100644
--- a/tests/storage/test_event_push_actions.py
+++ b/tests/storage/test_event_push_actions.py
@@ -146,12 +146,12 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
         _assert_counts(0, 0)
         _inject_actions(1, PlAIN_NOTIF)
         _assert_counts(1, 0)
-        _rotate(2)
+        _rotate(1)
         _assert_counts(1, 0)
 
         _inject_actions(3, PlAIN_NOTIF)
         _assert_counts(2, 0)
-        _rotate(4)
+        _rotate(3)
         _assert_counts(2, 0)
 
         _inject_actions(5, PlAIN_NOTIF)
@@ -162,7 +162,7 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
         _assert_counts(0, 0)
 
         _inject_actions(6, PlAIN_NOTIF)
-        _rotate(7)
+        _rotate(6)
         _assert_counts(1, 0)
 
         self.get_success(
@@ -178,13 +178,13 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
 
         _inject_actions(8, HIGHLIGHT)
         _assert_counts(1, 1)
-        _rotate(9)
+        _rotate(8)
         _assert_counts(1, 1)
 
         # Check that adding another notification and rotating after highlight
         # works.
         _inject_actions(10, PlAIN_NOTIF)
-        _rotate(11)
+        _rotate(10)
         _assert_counts(2, 1)
 
         # Check that sending read receipts at different points results in the