Only store the current value in the DB, not in the cache

Signed-off-by: Aurélien Bompard <aurelien@bompard.org>

fedbadges/cached.py

@@ -1,16 +1,13 @@
 import logging
 
 import pymemcache
-import redis
 from dogpile.cache import make_region
 from dogpile.cache.proxy import ProxyBackend
 
 
 log = logging.getLogger(__name__)
 cache = make_region()
 
-VERY_LONG_EXPIRATION_TIME = 86400 * 365  # a year
-
 
 def configure(**kwargs):
     if not cache.is_configured:
@@ -27,30 +24,3 @@ def set(self, key, value):
             if length == 2:
                 length = len(value[1])
             log.exception("Could not set the value in the cache (len=%s)", length)
-
-
-def get_cached_messages_count(badge_id: str, candidate: str, get_previous_fn):
-    key = f"messages_count|{badge_id}|{candidate}"
-    try:
-        current_value = cache.get_or_create(
-            key,
-            creator=lambda c: get_previous_fn(c) - 1,
-            creator_args=((candidate,), {}),
-            expiration_time=VERY_LONG_EXPIRATION_TIME,
-        )
-    except redis.exceptions.LockNotOwnedError:
-        # This happens when the get_previous_fn() call lasted longer that the maximum redis lock
-        # time. In this case, the value has been computed and stored in Redis, but dogpile just
-        # failed to release the lock. Just try again, the value should be there already.
-        log.debug(
-            "Oops, getting the cached messages count of rule %s for user %s took too long for the "
-            "cache lock. Retrying. It should be fast now.",
-            badge_id,
-            candidate,
-        )
-        return get_cached_messages_count(badge_id, candidate, get_previous_fn)
-
-    # Add one (the current message), store it, return it
-    new_value = current_value + 1
-    cache.set(key, new_value)
-    return new_value

fedbadges/rules.py

@@ -11,14 +11,16 @@
 import functools
 import inspect
 import logging
+import time
 from itertools import chain
 
 import datanommer.models
+import redis
+from dogpile.lock import Lock, NeedRegenerationException
 from fedora_messaging.api import Message
-from sqlalchemy.exc import IntegrityError
 from tahrir_api.dbapi import TahrirDatabase
 
-from fedbadges.cached import cache, get_cached_messages_count
+from fedbadges.cached import cache
 from fedbadges.fas import distgit2fas, krb2fas, openid2fas
 from fedbadges.utils import (
     # These are all in-process utilities
@@ -248,28 +250,49 @@ def _get_candidates(self, msg: Message, tahrir: TahrirDatabase):
         return candidates
 
     def _get_current_value(self, candidate: str, previous_count_fn, tahrir: TahrirDatabase):
-        candidate_email = f"{candidate}@{self.config['email_domain']}"
-        # First: the database
-        messages_count = tahrir.get_current_value(self.badge_id, candidate_email)
-        # If not found, use the cache
-        if messages_count is None:
-            # When we drop this cache sometime in the future, remember to keep the
-            # distributed lock (dogpile.lock) when running previous_count_fn()
-            messages_count = get_cached_messages_count(self.badge_id, candidate, previous_count_fn)
-        else:
-            # Found in DB! Add one (the current message)
-            messages_count += 1
-        # Store the value in the DB for next time
+        lock_key = f"messages_count|{self.badge_id}|{candidate}"
+        user_email = f"{candidate}@{self.config['email_domain']}"
+
+        def get_value():
+            value = tahrir.get_current_value(self.badge_id, user_email)
+            if value is None:
+                raise NeedRegenerationException()
+            # The dogpile.lock API ask us to return a (value, creation time) tuple
+            return value, time.time()
+
+        def gen_value():
+            # Ask Datanommer. We'll add the current message later, but it's already in DN,
+            # so substract it now.
+            value = previous_count_fn(candidate) - 1
+            # The dogpile.lock API ask us to return a (value, creation time) tuple
+            return value, time.time()
+
+        # Use a distributed lock to avoid two consumers processing the same rule for the same
+        # candidate at the same time and overwriting each other's values.
         try:
-            tahrir.set_current_value(self.badge_id, candidate_email, messages_count)
-            tahrir.session.commit()
-        except IntegrityError:
-            log.debug("Oops, conflict when setting the current value, let's try again")
-            # Another process already added the value! (querying datanommer can be long)
-            tahrir.session.rollback()
-            # Try again, this time the value should be in the DB already.
+            with Lock(
+                cache.backend.get_mutex(lock_key),
+                gen_value,
+                get_value,
+                expiretime=None,
+            ) as value:
+                # Add one (the current message)
+                new_value = value + 1
+                # Store the value in the DB for next time
+                tahrir.set_current_value(self.badge_id, user_email, new_value)
+                tahrir.session.commit()
+                return new_value
+        except redis.exceptions.LockNotOwnedError:
+            # This happens when the get_previous_fn() call lasted longer that the maximum redis lock
+            # time. In this case, the value has been computed and stored in the DB, but dogpile just
+            # failed to release the lock. Just try again, the value should be there already.
+            log.debug(
+                "Oops, getting the cached messages count of rule %s for user %s took too long for "
+                "the cache lock. Retrying. It should be fast now.",
+                self.badge_id,
+                candidate,
+            )
             return self._get_current_value(candidate, previous_count_fn, tahrir)
-        return messages_count
 
     def matches(self, msg: Message, tahrir: TahrirDatabase):
         # First, do a lightweight check to see if the msg matches a pattern.

tests/test_complicated_recipient.py

@@ -19,8 +19,8 @@ def user_exists(fasjson_client):
 
 @pytest.fixture
 def above_threshold():
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = float("inf")
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = float("inf")
         yield
 
 

tests/test_complicated_trigger.py

@@ -69,6 +69,6 @@ def count(self):
             return float("inf")  # Master tagger
 
     fasjson_client.get_user.return_value = SimpleNamespace(result={"username": "dummy-user"})
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = float("inf")
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = float("inf")
         assert rule.matches(message, tahrir_client) == {"ralph"}

tests/test_rule_matching.py

@@ -69,8 +69,8 @@ def test_full_simple_success(fasproxy, tahrir_client, fm_config, user_exists):
 
     msg = example_real_bodhi_message
 
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = 1
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = 1
         assert rule.matches(msg, tahrir_client) == {"lmacken"}
 
 
@@ -102,8 +102,8 @@ def test_full_simple_match_almost_succeed(fasproxy, tahrir_client, fm_config):
     # we should *fail* the ``matches`` call.
     msg = Message(topic="org.fedoraproject.prod.bodhi.mashtask.complete", body={"success": False})
 
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = 0
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = 0
         assert rule.matches(msg, tahrir_client) == set()
 
 
@@ -148,8 +148,8 @@ def test_yaml_specified_awardee_success(fasproxy, tahrir_client, fm_config, user
         },
     )
 
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = 1
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = 1
         assert rule.matches(msg, tahrir_client) == {"toshio", "ralph"}
 
 
@@ -187,8 +187,8 @@ def test_yaml_specified_awardee_failure(fasproxy, tahrir_client, fm_config, user
         },
     )
 
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = 1
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = 1
         assert rule.matches(msg, tahrir_client) == {"toshio"}
 
 
@@ -234,8 +234,8 @@ def test_against_duplicates(fasproxy, tahrir_client, fm_config, user_exists):
         },
     )
 
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = 1
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = 1
         assert rule.matches(msg, tahrir_client) == set(["ralph"])
 
 
@@ -269,8 +269,8 @@ def test_github_awardee(fasproxy, tahrir_client, fasjson_client, fm_config, user
         body={"user": "https://api.github.com/users/dummygh"},
     )
 
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = 1
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = 1
         fasjson_client.search.return_value = SimpleNamespace(result=[{"username": "dummy"}])
         assert rule.matches(msg, tahrir_client) == set(["dummy"])
         fasjson_client.search.assert_called_once_with(
@@ -336,6 +336,6 @@ def test_krb_awardee(fasproxy, tahrir_client, fm_config, user_exists):
         body={"owner": "packagerbot/os-master02.iad2.fedoraproject.org"},
     )
 
-    with patch("fedbadges.rules.get_cached_messages_count") as get_cached_messages_count:
-        get_cached_messages_count.return_value = 1
+    with patch("fedbadges.rules.BadgeRule._get_current_value") as get_current_value:
+        get_current_value.return_value = 1
         assert rule.matches(msg, tahrir_client) == set(["packagerbot"])