fix: keep the comment reader reading after a bogus head comment ID

AGENTS.md

@@ -258,7 +258,7 @@ A banned target (a 404 whose body says `"reason": "banned"`, `SUBREDDIT_IS_BANNE
 | [`scripts/negative_feedback_report.py`](./scripts/negative_feedback_report.py) | Writes an HTML report to `reports/negative_feedback/` (gitignored) of the "bad bot" exchanges with written feedback: a reply below a bot's `RESPOND_TO_NEGATIVE_SENTIMENT` question (or the "bad bot" comment itself) that says more than a bare sentiment phrase (bodies `[deleted]`, `[removed]` and `[unavailable]` never count). Each qualifying exchange shows only its new feedback, with the post and the Reddit-style comment tree collapsed behind a "Show full conversation" button. Scans all 4 bots by default (`--bot` picks some). Each report embeds the feedback ids it shows, so the next run only reports exchanges with new feedback, and links to the earlier reports. Reddit listings stop at about 1000 items, so very old comments may be unreachable. |
 | [`tests/conftest.py`](./tests/conftest.py) | `MockComment` / `MockSubmission` / `MockSubreddit` / `MockRedditor` plus the `make_mock_comment` and `make_mock_submission` factory fixtures. |
 | [`tests/unit/`](./tests/unit/) | 1700+ mocked unit tests (`@pytest.mark.unit`), one file per module. `test_crosspost_reply_phrasings.py` renders the crosspost reply for all 1330 combinations of locale, author, comment age and media-scan outcome (none, or each image and video scan) and checks each; its `render_crosspost_reply` is handy for reading them. |
-| [`tests/integration/test_reddit_integration.py`](./tests/integration/test_reddit_integration.py) | 10 **live** Reddit tests (`@pytest.mark.integration`) that submit real posts and comments. |
+| [`tests/integration/test_reddit_integration.py`](./tests/integration/test_reddit_integration.py) | 15 **live** Reddit tests (`@pytest.mark.integration`) that submit real posts and comments. |
 | [`tests/assets/test_sample.png`](./tests/assets/test_sample.png) | Base image for perceptual-hash variation tests. |
 | [`pytest.ini`](./pytest.ini) | Test discovery: `testpaths = tests`, `pythonpath = src .`, markers `unit` and `integration`, and `addopts = -m "not integration"`. |
 | [`.github/workflows/tests.yml`](./.github/workflows/tests.yml) | CI: `uv sync --locked` then `uv run pytest` on Python 3.11 (from `.python-version`), for pushes to `master`/`main` and for pull requests. Fails if `uv.lock` is out of date. On `master`, a `release` job then bumps the version and tags it (section 5), and, when the `AUTO_DEPLOY` variable is `true`, an `integration-tests` job runs the live tests with the bot credentials from repository secrets, then deletes what they created with `delete_test_artifacts.py --strict` (even after failed or cancelled tests), after which a `deploy` job redeploys the VM at the released commit ([`docs/auto-deploy.md`](./docs/auto-deploy.md)). |
@@ -467,8 +467,10 @@ uv run pytest -m integration
   - `reads`: ID blocks (or listing batches) fetched, and how many comments each returned (all of them, before
     the r/all filter). With `STREAM_SOURCE=listing`, about 100 comments/request means the listing returned
     full pages, so the bot was falling behind r/all.
-  - `head`: polls for the newest comment ID, and how many of them got an empty r/all listing. Empty polls
-    cost nothing but a few seconds of delay; only a long streak, which also logs a WARNING, stops new reads.
+  - `head`: polls for the newest comment ID, and how many of them got an empty r/all listing or were
+    `rejected` as implausible (see section 7). Empty polls cost nothing but a few seconds of delay; only a
+    long streak, which also logs a WARNING, stops new reads.
+  - If head polls succeeded but `reads: 0 requests`, a second line, a WARNING, says the reader is stalled.
   - `behind`: age of the newest comment read, and (`ids` only) how many IDs wait to be read. Growing values
     mean the readers do not keep up; `N ID block(s) failed` and `N IDs skipped` count comments given up on.
 
@@ -632,6 +634,14 @@ the blocks already done above it, so a restart neither skips nor re-reads them.
 `MAX_BACKLOG_IDS` (200,000, about half an hour at a guessed 100 comments a second) behind starts at the newest comment instead, and so does
 a running reader that falls that far behind (the skipped IDs are logged and counted).
 
+The head is checked before it is followed, because on 2026-10-17 a single bogus ID (about 6 × 10^18, real
+IDs were about 8 × 10^10) made the reader skip ahead to it and then read nothing for hours:
+- `fetch_head` ignores IDs in the listing more than `MAX_BACKLOG_IDS` from the median (`newest_consistent_id`).
+- A head more than `MAX_BACKLOG_IDS` plus `MAX_COMMENT_IDS_PER_SECOND` (2,000) per second since the last
+  accepted head (or the saved cursor, at startup) is rejected, logged once per streak and counted.
+- If `HEADS_BELOW_POSITION_BEFORE_RESET` (6) heads in a row are more than `MAX_BACKLOG_IDS` below the reader's
+  position, it restarts at the newest comment. A saved cursor ahead of the head is ignored the same way.
+
 **Reading comments (`STREAM_SOURCE=listing`).** `main_loop()` reads one batch from PRAW's comment stream. The
 stream uses `pause_after=-1`. In PRAW this makes the generator yield `None` after **every** request and
 **never sleep**, so the loop polls continuously, throttled only by prawcore's rate limiter. After a stream

changelog.d/+bogus-head-comment-id.fixed.md

@@ -0,0 +1,6 @@
+**The comment reader no longer stops reading after a bogus newest-comment ID.** On 2026-10-17 the r/all
+listing returned a comment ID about 6 × 10^18 (real IDs are about 8 × 10^10); the reader skipped ahead to it,
+never saw a higher head again, and scanned nothing for hours. Head IDs far from the rest of the listing, or
+implausibly far past the last accepted head, are now ignored (counted as `rejected` in the heartbeat), a
+reader position far ahead of every head resets itself after 6 polls, and the heartbeat logs a WARNING when
+heads were found but nothing was read. Deploying this also recovers a reader stuck on such a cursor.

src/racb/comment_reader.py

@@ -24,6 +24,7 @@
 comment, so their comments are now read too.
 """
 
+import datetime
 import logging
 import queue
 import threading
@@ -56,6 +57,17 @@
 # works on comments too old to be worth a reply. The same limit decides whether a restart resumes.
 MAX_BACKLOG_IDS = 200_000
 
+# How fast the head may plausibly move. Reddit hands out roughly 100 comment IDs a second
+# (scripts/verify_comment_ids.py); this allows 20 times that. On 2026-10-17 the r/all listing started returning
+# a comment whose ID was about 6 * 10^18, while real IDs were about 8 * 10^10: the reader jumped to it, never
+# saw a real head above its position again, and read nothing for hours. A head further ahead of the last
+# accepted one than MAX_BACKLOG_IDS plus this rate allows is now rejected instead of followed
+MAX_COMMENT_IDS_PER_SECOND = 2_000
+
+# Head polls in a row whose head is far below the reader's position before the reader trusts the head again.
+# A position that far ahead of every head can only come from a bogus ID, and it would stall reading for good
+HEADS_BELOW_POSITION_BEFORE_RESET = 6
+
 # Tries per block before its comments are given up on
 MAX_BLOCK_ATTEMPTS = 3
 
@@ -86,6 +98,13 @@ def number_to_comment_id(number):
     return ''.join(reversed(digits))
 
 
+def saved_timestamp(updated_at):
+    """The Unix time of a cursor's `updated_at`, a naive timestamp in the database's time zone (UTC)."""
+    if updated_at.tzinfo is None:
+        updated_at = updated_at.replace(tzinfo=datetime.timezone.utc)
+    return updated_at.timestamp()
+
+
 def block_fullnames(block_start):
     """The 100 comment fullnames of the block that starts at ID `block_start`."""
     return [f't1_{number_to_comment_id(number)}' for number in range(block_start, block_start + BLOCK_SIZE)]
@@ -112,9 +131,32 @@ def fetch_head():
     Returns None if the listing came back empty, which Reddit sometimes does even with HEAD_LISTING_LIMIT.
     """
     newest = list(reddit.get_reddit_instance().subreddit('all').comments(limit=HEAD_LISTING_LIMIT))
-    if not newest:
+    return newest_consistent_id([comment.id for comment in newest])
+
+
+def newest_consistent_id(comment_ids):
+    """Returns the highest of `comment_ids` (as a number) that lies near most of the others, or None.
+
+    The listing's comments were all written in the last few seconds, so their IDs lie within a few thousand
+    of each other. An ID far from the median (another numbering, or a malformed ID) is not a head to follow.
+    """
+    numbers = []
+    for comment_id in comment_ids:
+        try:
+            numbers.append(comment_id_to_number(comment_id))
+        except (AttributeError, TypeError, ValueError):
+            logger.debug(f'Ignoring comment ID {comment_id!r} in the r/all listing: not base 36')
+    if not numbers:
         return None
-    return max(comment_id_to_number(comment.id) for comment in newest)
+    median = sorted(numbers)[len(numbers) // 2]
+    consistent = [number for number in numbers if abs(number - median) <= MAX_BACKLOG_IDS]
+    if len(consistent) < len(numbers):
+        outliers = sorted(set(numbers) - set(consistent))
+        logger.warning(
+            f'Ignoring {len(numbers) - len(consistent)} of {len(numbers)} comment IDs in the r/all listing far '
+            f'from the others (median {number_to_comment_id(median)}, e.g. {number_to_comment_id(outliers[-1])})'
+        )
+    return max(consistent)
 
 
 def fetch_block(bot_name, block_start):
@@ -148,6 +190,12 @@ def __init__(self, handle_comment, reader_count, metrics=None, handle_exception=
         self.last_saved = time.monotonic()
         # Head polls in a row that got an empty listing, so a long streak is logged once, not every poll
         self.empty_head_polls = 0
+        # The last head accepted (or the saved cursor) and when (time.time()), to reject implausible jumps
+        self.reference_id = None
+        self.reference_time = None
+        # Head polls in a row rejected as too far ahead, and in a row far below the reader's position
+        self.rejected_heads = 0
+        self.heads_below_position = 0
         # Readers take turns between the accounts, so each account's rate limit carries a share
         self.threads = [
             threading.Thread(
@@ -201,15 +249,84 @@ def poll_head(self):
         with self._lock:
             if self.next_id is None:
                 self._choose_start(head)
+            if not self._accept_head(head):
+                return
+            self._reset_if_far_ahead_of(head)
             self._skip_ahead_if_too_far_behind(head)
             self._queue_blocks_up_to(head)
             lag_ids = head + 1 - self._watermark()
         if self.metrics:
             self.metrics.set_lag_ids(lag_ids)
 
+    def _max_plausible_head(self, now):
+        """The highest head that could be real, given the last accepted one; None if there is none yet."""
+        if self.reference_id is None:
+            return None
+        elapsed = max(0.0, now - self.reference_time)
+        return self.reference_id + MAX_BACKLOG_IDS + int(MAX_COMMENT_IDS_PER_SECOND * elapsed)
+
+    def _accept_head(self, head):
+        """False, with a WARNING once per streak, if `head` is too far ahead to be real. Call with the lock held."""
+        now = time.time()
+        max_head = self._max_plausible_head(now)
+        if max_head is not None and head > max_head:
+            self.rejected_heads += 1
+            if self.metrics:
+                self.metrics.record_rejected_head()
+            if self.rejected_heads == 1:
+                logger.warning(
+                    f'Ignoring head comment ID {number_to_comment_id(head)} ({head:,}): more than '
+                    f'{max_head - self.reference_id:,} IDs past the last one accepted, '
+                    f'{number_to_comment_id(self.reference_id)}'
+                )
+            return False
+        if self.rejected_heads:
+            logger.info(f'Accepted a head comment ID again after {self.rejected_heads} rejected one(s)')
+            self.rejected_heads = 0
+        if self.reference_id is None or head >= self.reference_id:
+            self.reference_id = head
+            self.reference_time = now
+        return True
+
+    def _reset_if_far_ahead_of(self, head):
+        """Moves the position back to `head` if every recent head was far below it. Call with the lock held."""
+        if self.next_id - (head + 1) <= MAX_BACKLOG_IDS:
+            self.heads_below_position = 0
+            return
+        self.heads_below_position += 1
+        if self.heads_below_position < HEADS_BELOW_POSITION_BEFORE_RESET:
+            return
+        logger.warning(
+            f'Comment reader position {number_to_comment_id(self.next_id)} is {self.next_id - head - 1:,} IDs '
+            f'ahead of the newest comment for {self.heads_below_position} polls; restarting at the newest comment'
+        )
+        self.next_id = head + 1
+        self.done_above.clear()
+        # The next head becomes the reference, in case this one was a bogus low ID
+        self.reference_id = None
+        self.reference_time = None
+        self.heads_below_position = 0
+
     def _choose_start(self, head):
-        """Resumes from the saved cursor if it is recent enough, otherwise starts at the head."""
+        """Resumes from the saved cursor if it is recent enough, otherwise starts at the head.
+
+        A saved cursor also bounds how far ahead the first head may plausibly be (see `_accept_head`), unless
+        it is ahead of the head itself, which only a bogus ID can have caused.
+        """
         saved = self._load_cursor()
+        if saved is not None and saved['next_comment_id'] <= head + 1 and saved.get('updated_at'):
+            self.reference_id = saved['next_comment_id']
+            self.reference_time = saved_timestamp(saved['updated_at'])
+            max_head = self._max_plausible_head(time.time())
+            if head > max_head:
+                # The first head is bogus: resume from the cursor and wait for a real head
+                logger.warning(
+                    f'Head comment ID {number_to_comment_id(head)} is implausibly far past the saved cursor '
+                    f'{number_to_comment_id(self.reference_id)}; resuming from the cursor'
+                )
+                self.next_id = saved['next_comment_id']
+                self.done_above = {start for start in saved['done_blocks'] or [] if start >= self.next_id}
+                return
         if saved is not None and 0 <= head + 1 - saved['next_comment_id'] <= MAX_BACKLOG_IDS:
             self.next_id = saved['next_comment_id']
             self.done_above = {start for start in saved['done_blocks'] or [] if start >= self.next_id}

src/racb/main.py

@@ -62,6 +62,7 @@ def __init__(self):
         self.interval_head_polls = 0
         self.interval_head_seconds = 0.0
         self.interval_empty_head_polls = 0
+        self.interval_rejected_heads = 0
         self.interval_failed_blocks = 0
         self.interval_skipped_ids = 0
         # Creation time of the newest comment read so far, and how many IDs wait to be read
@@ -104,6 +105,11 @@ def record_head_poll(self, seconds, empty=False):
             if empty:
                 self.interval_empty_head_polls += 1
 
+    def record_rejected_head(self):
+        """Counts a head comment ID ignored because it was implausibly far ahead (see comment_reader)."""
+        with self._lock:
+            self.interval_rejected_heads += 1
+
     def record_failed_block(self):
         """Counts a block of comment IDs given up on after repeated errors."""
         with self._lock:
@@ -135,6 +141,10 @@ def log_heartbeat(self):
             if self.interval_dropped:
                 message += f' | {self.interval_dropped} match(es) dropped, Phase 1 queue full'
             message += self._reading_summary(now)
+            # Heads came back but nothing was read: the reader is stuck, as on 2026-10-17
+            stalled = (
+                self.interval_head_polls > self.interval_empty_head_polls and self.interval_requests == 0
+            )
             self.interval_scanned = 0
             self.interval_matches = 0
             self.interval_aux = 0
@@ -145,10 +155,15 @@ def log_heartbeat(self):
             self.interval_head_polls = 0
             self.interval_head_seconds = 0.0
             self.interval_empty_head_polls = 0
+            self.interval_rejected_heads = 0
             self.interval_failed_blocks = 0
             self.interval_skipped_ids = 0
             self.last_log_time = now
         logger.info(message)
+        if stalled:
+            logger.warning(
+                '[Heartbeat] The newest comment ID was found but no comments were read: the comment reader is stalled'
+            )
 
     def _reading_summary(self, now):
         """The heartbeat's second half: how comments were read, and how far behind."""
@@ -162,7 +177,12 @@ def _reading_summary(self, now):
         else:
             parts.append('reads: 0 requests')
         if self.interval_head_polls:
-            empty = f' ({self.interval_empty_head_polls:,} empty)' if self.interval_empty_head_polls else ''
+            details = []
+            if self.interval_empty_head_polls:
+                details.append(f'{self.interval_empty_head_polls:,} empty')
+            if self.interval_rejected_heads:
+                details.append(f'{self.interval_rejected_heads:,} rejected')
+            empty = f' ({", ".join(details)})' if details else ''
             parts.append(
                 f'head: {self.interval_head_polls:,} polls{empty}, '
                 f'{self.interval_head_seconds / self.interval_head_polls:.2f} s/poll'