From eeb485f89447d246f90ae016d4e927275bbda5f6 Mon Sep 17 00:00:00 2001
From: Alex Vandiver <alexmv@zulip.com>
Date: Fri, 16 May 2025 14:05:49 +0000
Subject: [PATCH] process_fts_updates: After getting notifies, process them
 all.

Every insert into `fts_update_log` triggers a `NOTIFY`; processing a
batch of 1000 for every NOTIFY that we get thus results in a lot of
extra processing of no rows.

Instead, we clear all notifies, and repeat the pattern of processing
rows until the batch comes up short.  This may still result in extra
wake-ups, in the event that a new NOTIFY occurs after we clear them,
and that new row is processed by us in `update_all_rows`.  However,
these wake-ups are necessary for correctness, and this change will
result in drastically fewer extra checks.
---
 .../files/postgresql/process_fts_updates      | 35 +++++++++++--------
 1 file changed, 21 insertions(+), 14 deletions(-)

diff --git a/puppet/zulip/files/postgresql/process_fts_updates b/puppet/zulip/files/postgresql/process_fts_updates
index f8eecfbb18..4da7c310f2 100755
--- a/puppet/zulip/files/postgresql/process_fts_updates
+++ b/puppet/zulip/files/postgresql/process_fts_updates
@@ -91,6 +91,23 @@ def update_fts_columns(conn: psycopg2.extensions.connection) -> int:
         return len(ids)
 
 
+def update_all_rows(msg: str, conn: psycopg2.extensions.connection) -> None:
+    while True:
+        start_time = time.perf_counter()
+        rows_updated = update_fts_columns(conn)
+        if rows_updated:
+            logger.log(
+                logging.INFO,
+                "process_fts_updates: %s %d rows, %d rows/sec",
+                msg,
+                rows_updated,
+                rows_updated / (time.perf_counter() - start_time),
+            )
+
+        if rows_updated != BATCH_SIZE:
+            return
+
+
 def am_master(conn: psycopg2.extensions.connection) -> bool:
     with conn.cursor() as cursor:
         cursor.execute("SELECT pg_is_in_recovery()")
@@ -223,24 +240,14 @@ while True:
                 conn.commit()
 
             # Catch up on any historical columns
-            while True:
-                rows_updated = update_fts_columns(conn)
-                logger.log(
-                    logging.INFO if rows_updated > 0 else logging.DEBUG,
-                    "process_fts_updates: Processed %d rows catching up",
-                    rows_updated,
-                )
-
-                if rows_updated != BATCH_SIZE:
-                    # We're caught up, so proceed to the listening for updates phase.
-                    break
+            update_all_rows("Caught up", conn)
 
         # TODO: If we go back into recovery, we should stop processing updates
         if select.select([conn], [], [], 30) != ([], [], []):
             conn.poll()
-            while conn.notifies:
-                conn.notifies.pop()
-                update_fts_columns(cursor)
+            conn.notifies.clear()
+            update_all_rows("Updated", conn)
+
     except psycopg2.OperationalError as e:
         retries -= 1
         if retries <= 0: