Add an HTTP API for removing rejected pushers.

When a push is rejected by the push gateway then synapse needs to
remove the pusher from the database. However we probably don't want
to do that directly from the slave, so we add an HTTP API to synapse
to remove the pusher from the database.

synapse/app/pusher.py

@@ -89,6 +89,18 @@ class PusherServer(HomeServer):
         self.datastore = PusherSlaveStore(self.get_db_conn(), self)
         logger.info("Finished setting up.")
 
+    def remove_pusher(self, app_id, push_key, user_id):
+        http_client = self.get_simple_http_client()
+        replication_url = self.config.replication_url
+        url = replication_url + "/remove_pushers"
+        return http_client.post_json_get_json(url, {
+            "remove": [{
+                "app_id": app_id,
+                "push_key": push_key,
+                "user_id": user_id,
+            }]
+        })
+
     @defer.inlineCallbacks
     def replicate(self):
         http_client = self.get_simple_http_client()

synapse/push/httppusher.py

@@ -230,7 +230,7 @@ class HttpPusher(object):
                         "Pushkey %s was rejected: removing",
                         pk
                     )
-                    yield self.hs.get_pusherpool().remove_pusher(
+                    yield self.hs.remove_pusher(
                         self.app_id, pk, self.user_id
                     )
         defer.returnValue(True)

synapse/replication/pusher_resource.py

@@ -0,0 +1,53 @@
+# Copyright 2016 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from synapse.http.server import respond_with_json_bytes, request_handler
+from synapse.http.servlet import parse_json_object_from_request
+
+from twisted.web.resource import Resource
+from twisted.web.server import NOT_DONE_YET
+from twisted.internet import defer
+
+
+class PusherResource(Resource):
+    """
+    HTTP endpoint for deleting rejected pushers
+    """
+
+    def __init__(self, hs):
+        Resource.__init__(self)  # Resource is old-style, so no super()
+
+        self.version_string = hs.version_string
+        self.store = hs.get_datastore()
+        self.notifier = hs.get_notifier()
+
+    def render_POST(self, request):
+        self._async_render_POST(request)
+        return NOT_DONE_YET
+
+    @request_handler
+    @defer.inlineCallbacks
+    def _async_render_POST(self, request):
+        content = parse_json_object_from_request(request)
+
+        for remove in content["remove"]:
+            yield self.store.delete_pusher_by_app_id_pushkey_user_id(
+                remove["app_id"],
+                remove["push_key"],
+                remove["user_id"],
+            )
+
+        self.notifier.on_new_replication_data()
+
+        respond_with_json_bytes(request, 200, "{}")

synapse/replication/resource.py

@@ -15,6 +15,7 @@
 
 from synapse.http.servlet import parse_integer, parse_string
 from synapse.http.server import request_handler, finish_request
+from synapse.replication.pusher_resource import PusherResource
 
 from twisted.web.resource import Resource
 from twisted.web.server import NOT_DONE_YET
@@ -102,8 +103,6 @@ class ReplicationResource(Resource):
     long-polling this replication API for new data on those streams.
     """
 
-    isLeaf = True
-
     def __init__(self, hs):
         Resource.__init__(self)  # Resource is old-style, so no super()
 
@@ -114,6 +113,8 @@ class ReplicationResource(Resource):
         self.typing_handler = hs.get_handlers().typing_notification_handler
         self.notifier = hs.notifier
 
+        self.putChild("remove_pushers", PusherResource(hs))
+
     def render_GET(self, request):
         self._async_render_GET(request)
         return NOT_DONE_YET

synapse/server.py

@@ -193,6 +193,9 @@ class HomeServer(object):
             **self.db_config.get("args", {})
         )
 
+    def remove_pusher(self, app_id, push_key, user_id):
+        return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
+
 
 def _make_dependency_method(depname):
     def _get(hs):