activitypub.inbox: store seen activity ids in memcache, short circuit on seen ids earlier

activitypub.py

@@ -33,6 +33,7 @@ from common import (
     error,
     host_url,
     LOCAL_DOMAINS,
+    memcache,
     PRIMARY_DOMAIN,
     PROTOCOL_DOMAINS,
     redirect_wrap,
@@ -995,6 +996,16 @@ def inbox(protocol=None, id=None):
         body = request.get_data(as_text=True)
         error(f"Couldn't parse body as non-empty JSON mapping: {body}", exc_info=True)
 
+    # are we already processing or done with this activity?
+    id = activity.get('id')
+    if id:
+        key = f'AP-id-{id}'
+        if memcache.get(key):
+            logger.info(f'Already seen this activity {id}')
+            return '', 204
+        memcache.set(key, 'seen', expire=60 * 60)  # 1 hour in seconds
+
+    # check actor, signature, auth
     type = activity.get('type')
     actor = as1.get_object(activity, 'actor')
     actor_id = actor.get('id')
@@ -1040,8 +1051,8 @@ def inbox(protocol=None, id=None):
         followee_url = unwrap(util.get_url(activity, 'object'))
         activity.setdefault('url', f'{follower_url}#followed-{followee_url}')
 
-    id = (activity.get('id')
-          or f'{actor_id}#{type}-{object.get("id", "")}-{util.now().isoformat()}')
+    if not id:
+        id = f'{actor_id}#{type}-{object.get("id", "")}-{util.now().isoformat()}'
     obj = Object.get_or_create(id=id, as2=unwrap(activity), authed_as=authed_as,
                                source_protocol=ActivityPub.LABEL)
     return create_task(queue='receive', obj=obj.key.urlsafe(), authed_as=authed_as)

common.py

@@ -20,6 +20,7 @@ from oauth_dropins.webutil import appengine_info
 from oauth_dropins.webutil.appengine_info import DEBUG
 from oauth_dropins.webutil import flask_util
 import pymemcache.client.base
+from pymemcache.test.utils import MockMemcacheClient
 
 logger = logging.getLogger(__name__)
 
@@ -89,10 +90,12 @@ RUN_TASKS_INLINE = False  # overridden by unit tests
 OLD_ACCOUNT_AGE = timedelta(days=14)
 
 if appengine_info.DEBUG:
+    memcache = MockMemcacheClient()
     global_cache = _InProcessGlobalCache()
 else:
-    global_cache = MemcacheCache(pymemcache.client.base.PooledClient(
-        '10.126.144.3', timeout=10, connect_timeout=10))  # seconds
+    memcache = pymemcache.client.base.PooledClient(
+        '10.126.144.3', timeout=10, connect_timeout=10)  # seconds
+    global_cache = MemcacheCache(memcache)
 
 
 def base64_to_long(x):

tests/testutil.py

@@ -203,7 +203,14 @@ from activitypub import ActivityPub, CONNEG_HEADERS_AS2_HTML
 import atproto
 from atproto import ATProto
 import common
-from common import PRIMARY_DOMAIN, PROTOCOL_DOMAINS, OTHER_DOMAINS, LOCAL_DOMAINS
+from common import (
+    global_cache,
+    LOCAL_DOMAINS,
+    memcache,
+    OTHER_DOMAINS,
+    PRIMARY_DOMAIN,
+    PROTOCOL_DOMAINS,
+)
 from web import Web
 from flask_app import app
 
@@ -250,8 +257,8 @@ class TestCase(unittest.TestCase, testutil.Asserts):
 
         self.router_client = router.app.test_client()
 
-        app.wsgi_app.kwargs['global_cache'].clear()
-        router.app.wsgi_app.kwargs['global_cache'].clear()
+        memcache.clear()
+        global_cache.clear()
 
         # clear datastore
         requests.post(f'http://{ndb_client.host}/reset')