fix: handle a matching comment only after Reddit's unmarked-edit window

AGENTS.md

@@ -51,8 +51,11 @@ auxiliary bot's PRAW instance (`get_comment_with_different_praw_instance`). All
 > recommendations and get a reply from `same_subreddit_bot` or `same_post_bot`. `sub_doesnt_exist_bot` is the
 > one bot that also answers replies to other comments. For those, Phase 1 only checks whether the target exists
 > (`get_posts_with_same_content(..., existence_only=True)`: the URL search, no duplicate scan).
-> Every bot requires the **whole** body to be the subreddit reference (the matching rule above). A comment
-> edited after the bot answered (e.g. `r/name` later extended with `\n\nWait...`) keeps that answer.
+> Every bot requires the **whole** body to be the subreddit reference (the matching rule above). Reddit lets an
+> author edit a comment for about 3 minutes without marking it as edited, and people often extend a bare
+> `r/name` that way (`r/name makes me so mad...`). So Phase 1 handles a match only `EDIT_GRACE_SECONDS`
+> (4 minutes) after the comment was written, and checks the text as it is then. A later edit (marked as
+> edited on Reddit) still keeps the bot's answer.
 
 ---
 
@@ -133,7 +136,8 @@ A banned target (a 404 whose body says `"reason": "banned"`, `SUBREDDIT_IS_BANNE
     (`TARGET_SUBREDDIT_CHANGED`) if it no longer equals the stored target, case-insensitively. This blocks
     bait-and-switch edits. Rows scraped before v2.3 have `target_subreddit = NULL` and skip this check.
   - The reader threads (or the stream thread) only run the regex (`check_pattern`, no I/O) and queue the comment ID of a match. A
-    Phase 1 worker thread fetches the comment again through its own PRAW instance (one request) and does all
+    Phase 1 worker thread waits until 4 minutes after the comment was written (`workers.EDIT_GRACE_SECONDS`,
+    see the note in section 1), fetches the comment again through its own PRAW instance (one request) and does all
     of this work, including the duplicate search and image downloads (section 7).
   - Auxiliary replies go through `phase1.reply_as_bot()`, which only logs when `DRY_RUN` is enabled.
   - Any error while handling one comment is caught in `main.handle_comment_safely()` and skips only that
@@ -649,7 +653,8 @@ error, the new stream continues after the newest comment seen (`continue_after_i
 starts with `skip_existing=True`.
 
 Either way, for each comment the reading thread only counts it, buffers its subreddit name and runs the
-regex. A match goes into the Phase 1 queue as a comment ID; when the queue is full, the match is dropped and
+regex. A match goes into the Phase 1 queue as a comment ID with the time it is due (`created_utc` plus
+`EDIT_GRACE_SECONDS`), and a worker that takes it waits until then; when the queue is full, the match is dropped and
 counted in the heartbeat, so reading never waits for the workers.
 
 Rules for thread safety:

changelog.d/+edit-grace-period.fixed.md

@@ -0,0 +1 @@
+**Replies to Comments Edited Right After Posting:** Reddit lets an author edit a comment for about 3 minutes without marking it as edited. A bare `r/name` answered at once was often extended that way (`r/trucklodytes makes me so mad...`), leaving `same_subreddit_bot`'s reply under a comment it should have ignored. Phase 1 now handles a matching comment 4 minutes after it was written, with its text as it is then, so the auxiliary bots' replies come about 4 minutes later than before.

src/racb/main.py

@@ -18,7 +18,7 @@
 from racb.comment_reader import CommentReader
 from racb.core import db, known_subreddits, reddit, shutdown
 from racb.phases import phase1, phase2, phase3, inbox, cleanup
-from racb.workers import PeriodicJob, Phase1Workers
+from racb.workers import EDIT_GRACE_SECONDS, PeriodicJob, Phase1Workers
 
 logger = logging.getLogger(__name__)
 
@@ -500,7 +500,8 @@ def dispatch_comment(comment, phase1_workers):
         # Seeing a comment in a subreddit proves it exists; this only buffers the name (no I/O)
         known_subreddits.record(comment.subreddit.display_name)
         if phase1.check_pattern(comment) is not None:
-            phase1_workers.submit(comment.id)
+            # Handled once the author can no longer edit it unnoticed (`workers.EDIT_GRACE_SECONDS`)
+            phase1_workers.submit(comment.id, not_before=comment.created_utc + EDIT_GRACE_SECONDS)
     except Exception as e:
         if env_bool('DEBUG', False):
             raise

src/racb/workers.py

@@ -9,6 +9,7 @@
 import logging
 import queue
 import threading
+import time
 
 import prawcore
 
@@ -28,6 +29,12 @@
 # take well under 1 MB) but a backstop: a queue that stays full means Phase 1 cannot keep up at all.
 PHASE1_QUEUE_SIZE = 1000
 
+# How long after a comment was written Phase 1 waits before handling it. Reddit lets an author edit a
+# comment for about 3 minutes without marking it as edited, so a bare "r/name" answered at once can later
+# read "r/name makes me so mad" above a reply that no longer fits. After this wait, the worker fetches the
+# comment again and Phase 1 checks its final text.
+EDIT_GRACE_SECONDS = 4 * 60
+
 
 def run_job_safely(job_func):
     """Runs a scheduled job, logging (instead of propagating) its errors.
@@ -124,10 +131,14 @@ def start(self):
             thread.start()
         return self
 
-    def submit(self, comment_id):
-        """Queues a comment for Phase 1 without blocking. Returns False if the queue is full."""
+    def submit(self, comment_id, not_before=0):
+        """Queues a comment for Phase 1 without blocking. Returns False if the queue is full.
+
+        A worker handles it no earlier than `not_before` (a Unix timestamp). Comments arrive roughly in the
+        order they were written, so waiting for the oldest one never holds back one that is due.
+        """
         try:
-            self.queue.put_nowait(comment_id)
+            self.queue.put_nowait((comment_id, not_before))
         except queue.Full:
             # Logged once when the queue fills and once when it has room again; the heartbeat counts drops
             with self._dropped_lock:
@@ -155,9 +166,12 @@ def work(self):
         try:
             while not shutdown.stop_requested():
                 try:
-                    comment_id = self.queue.get(timeout=POLL_SECONDS)
+                    comment_id, not_before = self.queue.get(timeout=POLL_SECONDS)
                 except queue.Empty:
                     continue
+                wait_seconds = not_before - time.time()
+                if wait_seconds > 0 and shutdown.stop_requested_within(wait_seconds):
+                    return
                 try:
                     self.handle_comment_id(comment_id)
                 except reddit.TRANSIENT_NETWORK_ERRORS as e:

tests/conftest.py

@@ -108,8 +108,10 @@ def __init__(
         link_id='t3_sub_123',
         distinguished=None,
         submission=None,
+        created_utc=None,
     ):
         self.id = comment_id
+        self.created_utc = created_utc
         self.body = body
         self.permalink = permalink
         self.subreddit = MockSubreddit(subreddit) if isinstance(subreddit, str) else subreddit

tests/unit/test_main.py

@@ -176,9 +176,11 @@ def test_dispatch_comment_queues_only_matching_comments(make_mock_comment):
     phase1_workers = MagicMock()
     scanned_before = metrics.total_scanned
     with patch('racb.main.known_subreddits.record') as mock_record:
-        dispatch_comment(make_mock_comment(comment_id='match', body='r/pics'), phase1_workers)
+        match = make_mock_comment(comment_id='match', body='r/pics', created_utc=1_700_000_000.0)
+        dispatch_comment(match, phase1_workers)
         dispatch_comment(make_mock_comment(comment_id='other', body='nice picture'), phase1_workers)
-    phase1_workers.submit.assert_called_once_with('match')
+    # Not before the author's unnoticed-edit window has passed
+    phase1_workers.submit.assert_called_once_with('match', not_before=1_700_000_000.0 + 240)
     assert mock_record.call_count == 2
     assert metrics.total_scanned == scanned_before + 2
 
@@ -274,7 +276,7 @@ def test_main_loop_stops_reading_the_stream_on_shutdown(make_mock_comment):
     from racb.main import main_loop
     from racb.core import shutdown
     phase1_workers = MagicMock()
-    comments = [make_mock_comment(comment_id='first'), make_mock_comment(comment_id='second')]
+    comments = [make_mock_comment(comment_id=comment_id, created_utc=0) for comment_id in ('first', 'second')]
 
     def stream():
         yield comments[0]
@@ -283,7 +285,8 @@ def stream():
 
     with patch('racb.main.known_subreddits.record'):
         main_loop(stream(), phase1_workers)
-    phase1_workers.submit.assert_called_once_with('first')
+    phase1_workers.submit.assert_called_once()
+    assert phase1_workers.submit.call_args.args == ('first',)
 
 
 @pytest.mark.unit
@@ -370,7 +373,7 @@ def test_main_loop_records_each_batch_as_one_request(make_mock_comment):
     from racb.main import main_loop, metrics
     newest_seen = {'fullname': None}
     requests_before = metrics.interval_requests
-    comments = [make_mock_comment(comment_id='first'), make_mock_comment(comment_id='second')]
+    comments = [make_mock_comment(comment_id=comment_id, created_utc=0) for comment_id in ('first', 'second')]
     with patch('racb.main.known_subreddits.record'):
         main_loop(iter(comments + [None]), MagicMock(), newest_seen)
     assert metrics.interval_requests == requests_before + 1

tests/unit/test_workers.py

@@ -180,3 +180,29 @@ def handle(comment_id):
     workers.threads[0].join(WAIT_SECONDS)
     assert shutdown.stop_requested()
     assert shutdown.fatal_error() is error
+
+
+
+@pytest.mark.unit
+def test_phase1_workers_wait_until_a_comment_is_due():
+    handled = []
+    workers = Phase1Workers(handled.append, worker_count=1).start()
+    due = time.time() + 0.5
+    assert workers.submit('late', not_before=due)
+    time.sleep(0.2)
+    assert handled == []
+    assert wait_until(lambda: handled == ['late'])
+    assert time.time() >= due
+    shutdown.request_stop()
+
+
+@pytest.mark.unit
+def test_phase1_worker_stops_while_waiting_for_a_comment():
+    handled = []
+    workers = Phase1Workers(handled.append, worker_count=1).start()
+    workers.submit('far_future', not_before=time.time() + 3600)
+    time.sleep(0.1)
+    shutdown.request_stop()
+    workers.threads[0].join(WAIT_SECONDS)
+    assert not workers.threads[0].is_alive()
+    assert handled == []