diff --git a/synapse/storage/event_push_actions.py b/synapse/storage/event_push_actions.py
index 808c9d22fc..52d2bd3ffb 100644
--- a/synapse/storage/event_push_actions.py
+++ b/synapse/storage/event_push_actions.py
@@ -168,19 +168,13 @@ class EventPushActionsStore(SQLBaseStore):
row = txn.fetchone()
notify_count = row[0] if row else 0
- summary_notif_count = self._simple_select_one_onecol_txn(
- txn,
- table="event_push_summary",
- keyvalues={
- "user_id": user_id,
- "room_id": room_id,
- },
- retcol="notif_count",
- allow_none=True,
- )
-
- if summary_notif_count:
- notify_count += summary_notif_count
+ txn.execute("""
+ SELECT notif_count FROM event_push_summary
+ WHERE room_id = ? AND user_id = ? AND stream_ordering > ?
+ """, (room_id, user_id, stream_ordering,))
+ rows = txn.fetchall()
+ if rows:
+ notify_count += rows[0][0]
# Now get the number of highlights
sql = (
@@ -645,12 +639,20 @@ class EventPushActionsStore(SQLBaseStore):
# We want to make sure that we only ever do this one at a time
self.database_engine.lock_table(txn, "event_push_summary")
+ old_rotate_stream_ordering = self._simple_select_one_onecol_txn(
+ txn,
+ table="event_push_summary_stream_ordering",
+ keyvalues={},
+ retcol="stream_ordering",
+ )
+
# We don't to try and rotate millions of rows at once, so we cap the
# maximum stream ordering we'll rotate before.
txn.execute("""
SELECT stream_ordering FROM event_push_actions
+ WHERE stream_ordering > ?
ORDER BY stream_ordering ASC LIMIT 1 OFFSET 50000
- """)
+ """, (old_rotate_stream_ordering,))
stream_row = txn.fetchone()
if stream_row:
offset_stream_ordering, = stream_row
@@ -662,6 +664,8 @@ class EventPushActionsStore(SQLBaseStore):
rotate_to_stream_ordering = self.stream_ordering_day_ago
caught_up = True
+ logger.info("Rotating notifications up to: %s", rotate_to_stream_ordering)
+
self._rotate_notifs_before_txn(txn, rotate_to_stream_ordering)
# We have caught up iff we were limited by `stream_ordering_day_ago`
@@ -695,6 +699,8 @@ class EventPushActionsStore(SQLBaseStore):
txn.execute(sql, (old_rotate_stream_ordering, rotate_to_stream_ordering,))
rows = txn.fetchall()
+ logger.info("Rotating notifications, handling %d rows", len(rows))
+
# If the `old.user_id` above is NULL then we know there isn't already an
# entry in the table, so we simply insert it. Otherwise we update the
# existing table.
@@ -726,6 +732,8 @@ class EventPushActionsStore(SQLBaseStore):
(old_rotate_stream_ordering, rotate_to_stream_ordering,)
)
+ logger.info("Rotating notifications, deleted %s push actions", txn.rowcount)
+
txn.execute(
"UPDATE event_push_summary_stream_ordering SET stream_ordering = ?",
(rotate_to_stream_ordering,)
|
