diff --git a/synapse/push/mailer.py b/synapse/push/mailer.py
index 4ccaf178ce..73580c1c6c 100644
--- a/synapse/push/mailer.py
+++ b/synapse/push/mailer.py
@@ -555,10 +555,12 @@ class Mailer(object):
else:
# If the reason room doesn't have a name, say who the messages
# are from explicitly to avoid, "messages in the Bob room"
+ room_id = reason["room_id"]
+
sender_ids = list(
{
notif_events[n["event_id"]].sender
- for n in notifs_by_room[reason["room_id"]]
+ for n in notifs_by_room[room_id]
}
)
diff --git a/synapse/push/push_rule_evaluator.py b/synapse/push/push_rule_evaluator.py
index a70e677281..11032491af 100644
--- a/synapse/push/push_rule_evaluator.py
+++ b/synapse/push/push_rule_evaluator.py
@@ -16,9 +16,11 @@
import logging
import re
+from typing import Pattern
from six import string_types
+from synapse.events import EventBase
from synapse.types import UserID
from synapse.util.caches import register_cache
from synapse.util.caches.lrucache import LruCache
@@ -56,18 +58,18 @@ def _test_ineq_condition(condition, number):
rhs = m.group(2)
if not rhs.isdigit():
return False
- rhs = int(rhs)
+ rhs_int = int(rhs)
if ineq == "" or ineq == "==":
- return number == rhs
+ return number == rhs_int
elif ineq == "<":
- return number < rhs
+ return number < rhs_int
elif ineq == ">":
- return number > rhs
+ return number > rhs_int
elif ineq == ">=":
- return number >= rhs
+ return number >= rhs_int
elif ineq == "<=":
- return number <= rhs
+ return number <= rhs_int
else:
return False
@@ -83,7 +85,13 @@ def tweaks_for_actions(actions):
class PushRuleEvaluatorForEvent(object):
- def __init__(self, event, room_member_count, sender_power_level, power_levels):
+ def __init__(
+ self,
+ event: EventBase,
+ room_member_count: int,
+ sender_power_level: int,
+ power_levels: dict,
+ ):
self._event = event
self._room_member_count = room_member_count
self._sender_power_level = sender_power_level
@@ -92,7 +100,7 @@ class PushRuleEvaluatorForEvent(object):
# Maps strings of e.g. 'content.body' -> event["content"]["body"]
self._value_cache = _flatten_dict(event)
- def matches(self, condition, user_id, display_name):
+ def matches(self, condition: dict, user_id: str, display_name: str) -> bool:
if condition["kind"] == "event_match":
return self._event_match(condition, user_id)
elif condition["kind"] == "contains_display_name":
@@ -106,7 +114,7 @@ class PushRuleEvaluatorForEvent(object):
else:
return True
- def _event_match(self, condition, user_id):
+ def _event_match(self, condition: dict, user_id: str) -> bool:
pattern = condition.get("pattern", None)
if not pattern:
@@ -134,7 +142,7 @@ class PushRuleEvaluatorForEvent(object):
return _glob_matches(pattern, haystack)
- def _contains_display_name(self, display_name):
+ def _contains_display_name(self, display_name: str) -> bool:
if not display_name:
return False
@@ -142,51 +150,52 @@ class PushRuleEvaluatorForEvent(object):
if not body:
return False
- return _glob_matches(display_name, body, word_boundary=True)
+ # Similar to _glob_matches, but do not treat display_name as a glob.
+ r = regex_cache.get((display_name, False, True), None)
+ if not r:
+ r = re.escape(display_name)
+ r = _re_word_boundary(r)
+ r = re.compile(r, flags=re.IGNORECASE)
+ regex_cache[(display_name, False, True)] = r
+
+ return r.search(body)
- def _get_value(self, dotted_key):
+ def _get_value(self, dotted_key: str) -> str:
return self._value_cache.get(dotted_key, None)
-# Caches (glob, word_boundary) -> regex for push. See _glob_matches
+# Caches (string, is_glob, word_boundary) -> regex for push. See _glob_matches
regex_cache = LruCache(50000)
register_cache("cache", "regex_push_cache", regex_cache)
-def _glob_matches(glob, value, word_boundary=False):
+def _glob_matches(glob: str, value: str, word_boundary: bool = False) -> bool:
"""Tests if value matches glob.
Args:
- glob (string)
- value (string): String to test against glob.
- word_boundary (bool): Whether to match against word boundaries or entire
+ glob
+ value: String to test against glob.
+ word_boundary: Whether to match against word boundaries or entire
string. Defaults to False.
-
- Returns:
- bool
"""
try:
- r = regex_cache.get((glob, word_boundary), None)
+ r = regex_cache.get((glob, True, word_boundary), None)
if not r:
r = _glob_to_re(glob, word_boundary)
- regex_cache[(glob, word_boundary)] = r
+ regex_cache[(glob, True, word_boundary)] = r
return r.search(value)
except re.error:
logger.warning("Failed to parse glob to regex: %r", glob)
return False
-def _glob_to_re(glob, word_boundary):
+def _glob_to_re(glob: str, word_boundary: bool) -> Pattern:
"""Generates regex for a given glob.
Args:
- glob (string)
- word_boundary (bool): Whether to match against word boundaries or entire
- string. Defaults to False.
-
- Returns:
- regex object
+ glob
+ word_boundary: Whether to match against word boundaries or entire string.
"""
if IS_GLOB.search(glob):
r = re.escape(glob)
@@ -219,7 +228,7 @@ def _glob_to_re(glob, word_boundary):
return re.compile(r, flags=re.IGNORECASE)
-def _re_word_boundary(r):
+def _re_word_boundary(r: str) -> str:
"""
Adds word boundary characters to the start and end of an
expression to require that the match occur as a whole word,
diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py
index 01789a9fb4..88d203aa44 100644
--- a/synapse/push/pusherpool.py
+++ b/synapse/push/pusherpool.py
@@ -15,11 +15,17 @@
# limitations under the License.
import logging
+from collections import defaultdict
+from threading import Lock
+from typing import Dict, Tuple, Union
from twisted.internet import defer
+from synapse.metrics import LaterGauge
from synapse.metrics.background_process_metrics import run_as_background_process
from synapse.push import PusherConfigException
+from synapse.push.emailpusher import EmailPusher
+from synapse.push.httppusher import HttpPusher
from synapse.push.pusher import PusherFactory
from synapse.util.async_helpers import concurrently_execute
@@ -47,7 +53,29 @@ class PusherPool:
self._should_start_pushers = _hs.config.start_pushers
self.store = self.hs.get_datastore()
self.clock = self.hs.get_clock()
- self.pushers = {}
+
+ # map from user id to app_id:pushkey to pusher
+ self.pushers = {} # type: Dict[str, Dict[str, Union[HttpPusher, EmailPusher]]]
+
+ # a lock for the pushers dict, since `count_pushers` is called from an different
+ # and we otherwise get concurrent modification errors
+ self._pushers_lock = Lock()
+
+ def count_pushers():
+ results = defaultdict(int) # type: Dict[Tuple[str, str], int]
+ with self._pushers_lock:
+ for pushers in self.pushers.values():
+ for pusher in pushers.values():
+ k = (type(pusher).__name__, pusher.app_id)
+ results[k] += 1
+ return results
+
+ LaterGauge(
+ name="synapse_pushers",
+ desc="the number of active pushers",
+ labels=["kind", "app_id"],
+ caller=count_pushers,
+ )
def start(self):
"""Starts the pushers off in a background process.
@@ -271,11 +299,12 @@ class PusherPool:
return
appid_pushkey = "%s:%s" % (pusherdict["app_id"], pusherdict["pushkey"])
- byuser = self.pushers.setdefault(pusherdict["user_name"], {})
- if appid_pushkey in byuser:
- byuser[appid_pushkey].on_stop()
- byuser[appid_pushkey] = p
+ with self._pushers_lock:
+ byuser = self.pushers.setdefault(pusherdict["user_name"], {})
+ if appid_pushkey in byuser:
+ byuser[appid_pushkey].on_stop()
+ byuser[appid_pushkey] = p
# Check if there *may* be push to process. We do this as this check is a
# lot cheaper to do than actually fetching the exact rows we need to
@@ -304,7 +333,9 @@ class PusherPool:
if appid_pushkey in byuser:
logger.info("Stopping pusher %s / %s", user_id, appid_pushkey)
byuser[appid_pushkey].on_stop()
- del byuser[appid_pushkey]
+ with self._pushers_lock:
+ del byuser[appid_pushkey]
+
yield self.store.delete_pusher_by_app_id_pushkey_user_id(
app_id, pushkey, user_id
)
|
