push notif: Send a batch of message IDs in one remove payload.

When a bunch of messages with active notifications are all read at
once -- e.g. by the user choosing to mark all messages, or all in a
stream, as read, or just scrolling quickly through a PM conversation
-- there can be a large batch of this information to convey.  Doing it
in a single GCM/FCM message is better for server congestion, and for
the device's battery.

The corresponding client-side logic is in zulip/zulip-mobile#3343 .

Existing clients today only understand one message ID at a time; so
accommodate them by sending individual GCM/FCM messages up to an
arbitrary threshold, with the rest only as a batch.

Also add an explicit test for this logic.  The existing tests
that happen to cause this function to run don't exercise the
last condition, so without a new test `--coverage` complains.

zerver/lib/actions.py

@@ -4016,16 +4016,29 @@ def do_mark_stream_messages_as_read(user_profile: UserProfile,
 
 def do_clear_mobile_push_notifications_for_ids(user_profile: UserProfile,
                                                message_ids: List[int]) -> None:
-    for user_message in UserMessage.objects.filter(
+    filtered_message_ids = list(UserMessage.objects.filter(
-            message_id__in=message_ids,
+        message_id__in=message_ids,
-            user_profile=user_profile).extra(
+        user_profile=user_profile,
-                where=[UserMessage.where_active_push_notification()]):
+    ).extra(
-        event = {
+        where=[UserMessage.where_active_push_notification()],
-            "user_profile_id": user_profile.id,
+    ).values_list('message_id', flat=True))
-            "message_id": user_message.message_id,
+
+    num_detached = settings.MAX_UNBATCHED_REMOVE_NOTIFICATIONS - 1
+    for message_id in filtered_message_ids[:num_detached]:
+        # Older clients (all clients older than 2019-02-13) will only
+        # see the first message ID in a given notification-message.
+        # To help them out, send a few of these separately.
+        queue_json_publish("missedmessage_mobile_notifications", {
             "type": "remove",
-        }
+            "user_profile_id": user_profile.id,
-        queue_json_publish("missedmessage_mobile_notifications", event)
+            "message_ids": [message_id],
+        })
+    if filtered_message_ids[num_detached:]:
+        queue_json_publish("missedmessage_mobile_notifications", {
+            "type": "remove",
+            "user_profile_id": user_profile.id,
+            "message_ids": filtered_message_ids[num_detached:],
+        })
 
 def do_update_message_flags(user_profile: UserProfile,
                             client: Client,

zerver/lib/push_notifications.py

@@ -605,18 +605,21 @@ def get_message_payload_gcm(
     return data, gcm_options
 
 def get_remove_payload_gcm(
-        user_profile: UserProfile, message_id: int,
+        user_profile: UserProfile, message_ids: List[int],
 ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
     '''A `remove` payload + options, for Android via GCM/FCM.'''
     gcm_payload = get_base_payload(user_profile.realm)
     gcm_payload.update({
         'event': 'remove',
-        'zulip_message_id': message_id,  # message_id is reserved for CCS
+        'zulip_message_ids': ','.join(str(id) for id in message_ids),
+        # Older clients (all clients older than 2019-02-13) look only at
+        # `zulip_message_id` and ignore `zulip_message_ids`.  Do our best.
+        'zulip_message_id': message_ids[0],
     })
     gcm_options = {'priority': 'normal'}
     return gcm_payload, gcm_options
 
-def handle_remove_push_notification(user_profile_id: int, message_id: int) -> None:
+def handle_remove_push_notification(user_profile_id: int, message_ids: List[int]) -> None:
     """This should be called when a message that had previously had a
     mobile push executed is read.  This triggers a mobile push notifica
     mobile app when the message is read on the server, to remove the
@@ -624,8 +627,9 @@ def handle_remove_push_notification(user_profile_id: int, message_id: int) -> No
 
     """
     user_profile = get_user_profile_by_id(user_profile_id)
-    message, user_message = access_message(user_profile, message_id)
+    user_messages = [access_message(user_profile, message_id)[1]
-    gcm_payload, gcm_options = get_remove_payload_gcm(user_profile, message_id)
+                     for message_id in message_ids]
+    gcm_payload, gcm_options = get_remove_payload_gcm(user_profile, message_ids)
 
     if uses_notification_bouncer():
         try:
@@ -644,8 +648,10 @@ def handle_remove_push_notification(user_profile_id: int, message_id: int) -> No
         if android_devices:
             send_android_push_notification(android_devices, gcm_payload, gcm_options)
 
-    user_message.flags.active_mobile_push_notification = False
+    for user_message in user_messages:
-    user_message.save(update_fields=["flags"])
+        # TODO make this O(1) queries... including access_message, above
+        user_message.flags.active_mobile_push_notification = False
+        user_message.save(update_fields=["flags"])
 
 @statsd_increment("push_notifications")
 def handle_push_notification(user_profile_id: int, missed_message: Dict[str, Any]) -> None:

zerver/tests/test_push_notifications.py

@@ -15,6 +15,7 @@ import uuid
 from django.test import override_settings
 from django.conf import settings
 from django.http import HttpResponse
+from django.db.models import F
 from django.utils.crypto import get_random_string
 from django.utils.timezone import utc as timezone_utc
 
@@ -34,7 +35,7 @@ from zerver.models import (
     Stream,
     Subscription,
 )
-from zerver.lib.actions import do_delete_messages
+from zerver.lib.actions import do_delete_messages, do_mark_stream_messages_as_read
 from zerver.lib.soft_deactivation import do_soft_deactivate_users
 from zerver.lib.push_notifications import (
     absolute_avatar_url,
@@ -725,12 +726,17 @@ class HandlePushNotificationTest(PushNotificationTest):
                            '.send_notifications_to_bouncer') as mock_send_android, \
                 mock.patch('zerver.lib.push_notifications.get_base_payload',
                            return_value={'gcm': True}):
-            handle_remove_push_notification(user_profile.id, message.id)
+            handle_remove_push_notification(user_profile.id, [message.id])
-            mock_send_android.assert_called_with(user_profile.id, {},
+            mock_send_android.assert_called_with(
-                                                 {'gcm': True,
+                user_profile.id,
-                                                  'event': 'remove',
+                {},
-                                                  'zulip_message_id': message.id},
+                {
-                                                 {'priority': 'normal'})
+                    'gcm': True,
+                    'event': 'remove',
+                    'zulip_message_ids': str(message.id),
+                    'zulip_message_id': message.id,
+                },
+                {'priority': 'normal'})
             user_message = UserMessage.objects.get(user_profile=self.user_profile,
                                                    message=message)
             self.assertEqual(user_message.flags.active_mobile_push_notification, False)
@@ -753,12 +759,16 @@ class HandlePushNotificationTest(PushNotificationTest):
                         '.send_android_push_notification') as mock_send_android, \
                 mock.patch('zerver.lib.push_notifications.get_base_payload',
                            return_value={'gcm': True}):
-            handle_remove_push_notification(self.user_profile.id, message.id)
+            handle_remove_push_notification(self.user_profile.id, [message.id])
-            mock_send_android.assert_called_with(android_devices,
+            mock_send_android.assert_called_with(
-                                                 {'gcm': True,
+                android_devices,
-                                                  'event': 'remove',
+                {
-                                                  'zulip_message_id': message.id},
+                    'gcm': True,
-                                                 {'priority': 'normal'})
+                    'event': 'remove',
+                    'zulip_message_ids': str(message.id),
+                    'zulip_message_id': message.id,
+                },
+                {'priority': 'normal'})
             user_message = UserMessage.objects.get(user_profile=self.user_profile,
                                                    message=message)
             self.assertEqual(user_message.flags.active_mobile_push_notification, False)
@@ -1518,6 +1528,39 @@ class GCMSendTest(PushNotificationTest):
         c1 = call("GCM: Delivery to %s failed: Failed" % (token,))
         mock_warn.assert_has_calls([c1], any_order=True)
 
+class TestClearOnRead(ZulipTestCase):
+    def test_mark_stream_as_read(self) -> None:
+        n_msgs = 3
+        max_unbatched = 2
+
+        hamlet = self.example_user("hamlet")
+        hamlet.enable_stream_push_notifications = True
+        hamlet.save()
+        stream = self.subscribe(hamlet, "Denmark")
+
+        msgids = [self.send_stream_message(self.example_email("iago"),
+                                           stream.name,
+                                           "yo {}".format(i))
+                  for i in range(n_msgs)]
+        UserMessage.objects.filter(
+            user_profile_id=hamlet.id,
+            message_id__in=msgids,
+        ).update(
+            flags=F('flags').bitor(
+                UserMessage.flags.active_mobile_push_notification))
+
+        with mock.patch("zerver.lib.actions.queue_json_publish") as mock_publish:
+            with override_settings(MAX_UNBATCHED_REMOVE_NOTIFICATIONS=max_unbatched):
+                do_mark_stream_messages_as_read(hamlet, self.client, stream)
+            queue_items = [c[0][1] for c in mock_publish.call_args_list]
+            groups = [item['message_ids'] for item in queue_items]
+
+        self.assertEqual(len(groups), min(len(msgids), max_unbatched))
+        for g in groups[:-1]:
+            self.assertEqual(len(g), 1)
+        self.assertEqual(sum(len(g) for g in groups), len(msgids))
+        self.assertEqual(set(id for g in groups for id in g), set(msgids))
+
 class TestReceivesNotificationsFunctions(ZulipTestCase):
     def setUp(self) -> None:
         self.user = self.example_user('cordelia')

zerver/worker/queue_processors.py

@@ -360,7 +360,10 @@ class PushNotificationsWorker(QueueProcessingWorker):  # nocoverage
 
     def consume(self, data: Mapping[str, Any]) -> None:
         if data.get("type", "add") == "remove":
-            handle_remove_push_notification(data['user_profile_id'], data['message_id'])
+            message_ids = data.get('message_ids')
+            if message_ids is None:  # legacy task across an upgrade
+                message_ids = [data['message_id']]
+            handle_remove_push_notification(data['user_profile_id'], message_ids)
         else:
             handle_push_notification(data['user_profile_id'], data)
 

zproject/settings.py

@@ -339,6 +339,13 @@ DEFAULT_SETTINGS.update({
     'APNS_CERT_FILE': None,
     'APNS_SANDBOX': True,
 
+    # Max number of "remove notification" FCM/GCM messages to send separately
+    # in one burst; the rest are batched.  Older clients ignore the batched
+    # portion, so only receive this many removals.  Lower values mitigate
+    # server congestion and client battery use.  To batch unconditionally,
+    # set to 1.
+    'MAX_UNBATCHED_REMOVE_NOTIFICATIONS': 10,
+
     # Limits related to the size of file uploads; last few in MB.
     'DATA_UPLOAD_MAX_MEMORY_SIZE': 25 * 1024 * 1024,
     'MAX_AVATAR_FILE_SIZE': 5,