Action deduplication in filtering system (#3533)

Author: ChrisLovering

bot/exts/filtering/filtering.py

@@ -2,6 +2,7 @@
 import io
 import json
 import re
+import time
 import unicodedata
 from collections import defaultdict
 from collections.abc import Iterable, Mapping
@@ -72,6 +73,7 @@
 OFFENSIVE_MSG_DELETE_TIME = datetime.timedelta(days=7)
 WEEKLY_REPORT_ISO_DAY = 3  # 1=Monday, 7=Sunday
 MAX_IMAGE_HASH_SIZE = 5_000_000
+ACTION_DEDUPE_WINDOW_SECONDS = 3
 
 
 def _clean_ban_mentions(mentions: set[str]) -> set[str]:
@@ -113,6 +115,7 @@ def __init__(self, bot: Bot):
         self.bot = bot
         self.filter_lists: dict[str, FilterList] = {}
         self._subscriptions = defaultdict[Event, list[FilterList]](list)
+        self._recent_actions: dict[tuple, float] = {}
         self.delete_scheduler = scheduling.Scheduler(self.__class__.__name__)
         self.webhook: discord.Webhook | None = None
 
@@ -267,8 +270,7 @@ async def on_message(self, msg: Message) -> None:
 
         result_actions, list_messages, triggers = await self._resolve_action(ctx)
         self.message_cache.update(msg, metadata=triggers)
-        if result_actions:
-            await result_actions.action(ctx)
+        await self._run_actions(ctx, result_actions)
         if ctx.send_alert:
             await self._send_alert(ctx, list_messages)
 
@@ -298,8 +300,7 @@ async def on_message_edit(self, before: discord.Message, after: discord.Message)
         self.message_cache.update(after)
         ctx = FilterContext.from_message(Event.MESSAGE_EDIT, after, before, self.message_cache)
         result_actions, list_messages, triggers = await self._resolve_action(ctx)
-        if result_actions:
-            await result_actions.action(ctx)
+        await self._run_actions(ctx, result_actions)
         if ctx.send_alert:
             await self._send_alert(ctx, list_messages)
         await self._maybe_schedule_msg_delete(ctx, result_actions)
@@ -334,8 +335,7 @@ async def filter_snekbox_output(
         ctx = FilterContext.from_message(Event.SNEKBOX, msg).replace(content=content, attachments=files)
 
         result_actions, list_messages, triggers = await self._resolve_action(ctx)
-        if result_actions:
-            await result_actions.action(ctx)
+        await self._run_actions(ctx, result_actions)
         if ctx.send_alert:
             await self._send_alert(ctx, list_messages)
 
@@ -1076,6 +1076,35 @@ async def _send_alert(self, ctx: FilterContext, triggered_filters: dict[FilterLi
             username=name, content=ctx.alert_content, embeds=[embed, *ctx.alert_embeds][:10], view=AlertView(ctx)
         )
 
+    async def _run_actions(self, ctx: FilterContext, actions: ActionSettings | None) -> None:
+        """Execute actions unless this exact action payload was run very recently for the same context source."""
+        if actions and self._should_run_actions(ctx, actions):
+            await actions.action(ctx)
+
+    def _should_run_actions(self, ctx: FilterContext, actions: ActionSettings) -> bool:
+        """Return whether actions should run, suppressing identical actions for the same source within a time window."""
+        now = time.monotonic()
+        recent = self._recent_actions
+        # Evict stale cache keys from the front.
+        while recent:
+            oldest_key = next(iter(recent))
+            if now - recent[oldest_key] < ACTION_DEDUPE_WINDOW_SECONDS:
+                break
+            del recent[oldest_key]
+
+        key = (
+            ctx.event.name,
+            getattr(ctx.author, "id", None),
+            getattr(ctx.channel, "id", None),
+            json.dumps(to_serializable(actions), sort_keys=True, default=str),
+            # So that each callable in additional_actions also forms the cache key
+            tuple(sorted(getattr(a, "__qualname__", repr(a)) for a in ctx.additional_actions)),
+        )
+        if key in recent:
+            return False
+        recent[key] = now
+        return True
+
     def _increment_stats(self, triggered_filters: dict[AtomicList, list[Filter]]) -> None:
         """Increment the stats for every filter triggered."""
         for filters in triggered_filters.values():
@@ -1116,8 +1145,7 @@ async def _check_bad_name(self, ctx: FilterContext) -> FilterContext:
         new_ctx = ctx.replace(content=" ".join(names_to_check))
         result_actions, list_messages, triggers = await self._resolve_action(new_ctx)
         new_ctx = new_ctx.replace(content=ctx.content)  # Alert with the original content.
-        if result_actions:
-            await result_actions.action(new_ctx)
+        await self._run_actions(new_ctx, result_actions)
         if new_ctx.send_alert:
             await self._send_alert(new_ctx, list_messages)
         self._increment_stats(triggers)

tests/bot/exts/filtering/test_action_deduplication.py

@@ -0,0 +1,51 @@
+import unittest
+
+from bot.exts.filtering._filter_context import Event, FilterContext
+from bot.exts.filtering._settings import ActionSettings
+from bot.exts.filtering._settings_types.actions.send_alert import SendAlert
+from bot.exts.filtering.filtering import Filtering
+from tests.helpers import MockBot, MockMember, MockTextChannel
+
+
+class ActionDeduplicationTests(unittest.TestCase):
+    def setUp(self):
+        self.cog = Filtering(MockBot())
+
+    @staticmethod
+    def _make_actions(send_alert: bool) -> ActionSettings:
+        actions = ActionSettings({})
+        actions[SendAlert.name] = SendAlert(send_alert=send_alert)
+        return actions
+
+    @staticmethod
+    def _make_context(*, author_id: int = 1, channel_id: int = 10) -> FilterContext:
+        return FilterContext(
+            event=Event.MESSAGE,
+            author=MockMember(id=author_id),
+            channel=MockTextChannel(id=channel_id),
+            content="hello",
+            message=None,
+        )
+
+    def test_identical_actions_in_window_are_suppressed(self):
+        ctx = self._make_context()
+        actions = self._make_actions(send_alert=True)
+
+        self.assertTrue(self.cog._should_run_actions(ctx, actions))
+        self.assertFalse(self.cog._should_run_actions(ctx, actions))
+
+    def test_different_action_values_are_not_suppressed(self):
+        ctx = self._make_context()
+        first_actions = self._make_actions(send_alert=True)
+        second_actions = self._make_actions(send_alert=False)
+
+        self.assertTrue(self.cog._should_run_actions(ctx, first_actions))
+        self.assertTrue(self.cog._should_run_actions(ctx, second_actions))
+
+    def test_same_action_payload_for_different_authors_is_not_suppressed(self):
+        first_ctx = self._make_context(author_id=1)
+        second_ctx = self._make_context(author_id=2)
+        actions = self._make_actions(send_alert=True)
+
+        self.assertTrue(self.cog._should_run_actions(first_ctx, actions))
+        self.assertTrue(self.cog._should_run_actions(second_ctx, actions))