push_notifs: Gracefully handle exception when server cant push.

The problem was that earlier this was just an uncaught JsonableError,
leading to a full traceback getting spammed to the admins.
The prior commit introduced a clear .code for this error on the bouncer
side, meaning the self-hosted server can now detect that and handle it
nicely, by just logging.error about it and also take the opportunity to
adjust the realm.push_notifications_... flags.

zerver/lib/push_notifications.py

@@ -610,7 +610,20 @@ def send_notifications_to_bouncer(
         "apple_devices": [device.token for device in apple_devices],
     }
     # Calls zilencer.views.remote_server_notify_push
+
+    try:
         response_data = send_json_to_push_bouncer("POST", "push/notify", post_data)
+    except PushNotificationsDisallowedByBouncerError as e:
+        logger.warning("Bouncer refused to send push notification: %s", e.reason)
+        do_set_realm_property(
+            user_profile.realm,
+            "push_notifications_enabled",
+            False,
+            acting_user=None,
+        )
+        do_set_push_notifications_enabled_end_timestamp(user_profile.realm, None, acting_user=None)
+        return
+
     assert isinstance(response_data["total_android_devices"], int)
     assert isinstance(response_data["total_apple_devices"], int)
 
@@ -1504,3 +1517,8 @@ class InvalidRemotePushDeviceTokenError(JsonableError):
     @override
     def msg_format() -> str:
         return _("Device not recognized by the push bouncer")
+
+
+class PushNotificationsDisallowedByBouncerError(Exception):
+    def __init__(self, reason: str) -> None:
+        self.reason = reason

zerver/lib/remote_server.py

@@ -185,6 +185,10 @@ def send_to_push_bouncer(
             raise PushNotificationBouncerError(
                 _("Push notifications bouncer error: {error}").format(error=msg)
             )
+        elif "code" in result_dict and result_dict["code"] == "PUSH_NOTIFICATIONS_DISALLOWED":
+            from zerver.lib.push_notifications import PushNotificationsDisallowedByBouncerError
+
+            raise PushNotificationsDisallowedByBouncerError(reason=msg)
         elif (
             endpoint == "push/test_notification"
             and "code" in result_dict

zerver/tests/test_push_notifications.py

@@ -2738,6 +2738,80 @@ class HandlePushNotificationTest(PushNotificationTest):
                 ),
             )
 
+    @override_settings(PUSH_NOTIFICATION_BOUNCER_URL="https://push.zulip.org.example.com")
+    @responses.activate
+    def test_end_to_end_failure_due_to_no_plan(self) -> None:
+        self.add_mock_response()
+
+        self.setup_apns_tokens()
+        self.setup_gcm_tokens()
+
+        self.server.last_api_feature_level = 237
+        self.server.save()
+
+        realm = self.user_profile.realm
+        realm.push_notifications_enabled = True
+        realm.save()
+
+        message = self.get_message(
+            Recipient.PERSONAL,
+            type_id=self.personal_recipient_user.id,
+            realm_id=self.personal_recipient_user.realm_id,
+        )
+        UserMessage.objects.create(
+            user_profile=self.user_profile,
+            message=message,
+        )
+
+        missed_message = {
+            "message_id": message.id,
+            "trigger": NotificationTriggers.DIRECT_MESSAGE,
+        }
+        with mock.patch(
+            "corporate.lib.stripe.RemoteServerBillingSession.current_count_for_billed_licenses",
+            return_value=100,
+        ) as mock_current_count, self.assertLogs(
+            "zerver.lib.push_notifications", level="INFO"
+        ) as pn_logger, self.assertLogs(
+            "zilencer.views", level="INFO"
+        ):
+            handle_push_notification(self.user_profile.id, missed_message)
+
+            self.assertEqual(
+                pn_logger.output,
+                [
+                    f"INFO:zerver.lib.push_notifications:Sending push notifications to mobile clients for user {self.user_profile.id}",
+                    "WARNING:zerver.lib.push_notifications:Bouncer refused to send push notification: Your plan doesn't allow sending push notifications. Reason provided by the server: No plan many users",
+                ],
+            )
+            realm.refresh_from_db()
+            self.assertEqual(realm.push_notifications_enabled, False)
+            self.assertEqual(realm.push_notifications_enabled_end_timestamp, None)
+
+            # Now verify the flag will correctly get flipped back if the server stops
+            # rejecting our notification.
+
+            # This will put us within the allowed number of users to use push notifications
+            # for free, so the server will accept our next request.
+            mock_current_count.return_value = 5
+
+            new_message_id = self.send_personal_message(
+                self.example_user("othello"), self.user_profile
+            )
+            new_missed_message = {
+                "message_id": new_message_id,
+                "trigger": NotificationTriggers.DIRECT_MESSAGE,
+            }
+
+            handle_push_notification(self.user_profile.id, new_missed_message)
+            self.assertIn(
+                f"Sent mobile push notifications for user {self.user_profile.id}",
+                pn_logger.output[-1],
+            )
+            realm.refresh_from_db()
+            self.assertEqual(realm.push_notifications_enabled, True)
+            self.assertEqual(realm.push_notifications_enabled_end_timestamp, None)
+
     @override_settings(PUSH_NOTIFICATION_BOUNCER_URL="https://push.zulip.org.example.com")
     @responses.activate
     def test_unregistered_client(self) -> None: