From 4eb0efcd4886950b9466524ffebc23c967865774 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 06:59:04 +0000 Subject: [PATCH] fix: keep the comment reader reading after a bogus head comment ID On 2026-10-17 the r/all listing returned a comment ID of about 6 * 10^18. The reader skipped ahead to it, never saw a higher head again and scanned nothing for hours. Outlier IDs in the listing and implausible head jumps are now ignored, a position far ahead of every head resets itself, and the heartbeat warns when heads were found but nothing was read. Adds unit and live regression tests. Co-Authored-By: Toldry Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019ddUNNyEKK7YQuAyfKfGwi --- AGENTS.md | 16 +- changelog.d/+bogus-head-comment-id.fixed.md | 6 + src/racb/comment_reader.py | 123 +++++++++- src/racb/main.py | 22 +- tests/integration/test_reddit_integration.py | 74 ++++++ tests/unit/test_comment_reader.py | 226 ++++++++++++++++++- tests/unit/test_main.py | 45 ++++ 7 files changed, 502 insertions(+), 10 deletions(-) create mode 100644 changelog.d/+bogus-head-comment-id.fixed.md diff --git a/AGENTS.md b/AGENTS.md index c7e69c5..8c69170 100644 --- a/AGENTS.md +++ b/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 diff --git a/changelog.d/+bogus-head-comment-id.fixed.md b/changelog.d/+bogus-head-comment-id.fixed.md new file mode 100644 index 0000000..a5531cf --- /dev/null +++ b/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. diff --git a/src/racb/comment_reader.py b/src/racb/comment_reader.py index 38e58df..98c453e 100644 --- a/src/racb/comment_reader.py +++ b/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} diff --git a/src/racb/main.py b/src/racb/main.py index 6aa3e68..ed7868c 100644 --- a/src/racb/main.py +++ b/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' diff --git a/tests/integration/test_reddit_integration.py b/tests/integration/test_reddit_integration.py index 070a030..694674e 100644 --- a/tests/integration/test_reddit_integration.py +++ b/tests/integration/test_reddit_integration.py @@ -3,6 +3,7 @@ import logging import os import tempfile +import threading import time import uuid from urllib.parse import urlparse @@ -133,6 +134,79 @@ def test_live_comment_reader_reads_recent_comments_by_id(): assert any(comment_reader.is_shown_on_r_all(comment) for comment in comments) +@pytest.mark.integration +def test_live_head_is_a_recent_real_comment(): + """The head the reader follows must be a real, recent comment in the numbering it reads. + + On 2026-10-17 the head became a bogus ID (about 6 * 10^18) and the reader read nothing for hours. The block + ending at the head must hold comments written in the last hour, with IDs no higher than the head. + """ + head = fetch_head_with_retries() + first = head - comment_reader.BLOCK_SIZE + 1 + comments = comment_reader.fetch_block(reddit.AUTO_CROSSPOST_BOT_NAME, first) + assert comments, f'no comment in the block ending at the head {comment_reader.number_to_comment_id(head)}' + numbers = [comment_reader.comment_id_to_number(comment.id) for comment in comments] + assert all(first <= number <= head for number in numbers) + newest_age = time.time() - max(comment.created_utc for comment in comments) + assert newest_age < 3600, f'the newest comment at the head is {newest_age:.0f} s old' + + +class InMemoryCursorStore: + """Keeps the live reader test's cursor out of the database: no saved cursor, saves are only counted.""" + + def __init__(self): + self.saves = 0 + + def get_stream_cursor(self, name): + return None + + def save_stream_cursor(self, name, next_comment_id, done_blocks): + self.saves += 1 + + +@pytest.mark.integration +def test_live_comment_reader_scans_comments_after_start(): + """Runs the real CommentReader against Reddit for about 45 s: it must read and hand on comments. + + This is the check the 2026-10-17 outage would have failed: the head polls succeeded, but the reader + made no read requests and scanned nothing. + """ + from racb.core import shutdown + from racb.main import StreamMetrics + + handled = [] + errors = [] + metrics = StreamMetrics() + reader = comment_reader.CommentReader( + handled.append, + reader_count=2, + metrics=metrics, + handle_exception=lambda error: errors.append(error) or False, + cursor_store=InMemoryCursorStore(), + ) + run_thread = threading.Thread(target=reader.run, name='live-reader-test', daemon=True) + reader.start() + run_thread.start() + try: + # Read for 45 s, and up to 90 s while fewer than 200 comments came in + started = time.monotonic() + while time.monotonic() - started < 45 or (len(handled) < 200 and time.monotonic() - started < 90): + time.sleep(1) + finally: + shutdown.request_stop() + run_thread.join(30) + for thread in reader.threads: + thread.join(30) + shutdown.reset() + assert metrics.interval_requests > 0, f'no read request; head poll errors: {errors}' + assert len(handled) >= 200, f'only {len(handled)} comments read; head poll errors: {errors}' + assert metrics.lag_ids is not None + # Slightly negative when the last head was a little older than the one before; the outage's was -6 * 10^18 + assert abs(metrics.lag_ids) < comment_reader.MAX_BACKLOG_IDS + newest_age = time.time() - metrics.newest_created_utc + assert newest_age < 600, f'the newest comment read is {newest_age:.0f} s old' + + @pytest.mark.integration def test_live_subreddits_access(): """Verifies that the bot can access the private test subreddits.""" diff --git a/tests/unit/test_comment_reader.py b/tests/unit/test_comment_reader.py index d2b9f93..3fcf954 100644 --- a/tests/unit/test_comment_reader.py +++ b/tests/unit/test_comment_reader.py @@ -13,11 +13,14 @@ from racb.comment_reader import ( BLOCK_SIZE, CURSOR_NAME, + HEADS_BELOW_POSITION_BEFORE_RESET, MAX_BACKLOG_IDS, + MAX_COMMENT_IDS_PER_SECOND, CommentReader, block_fullnames, comment_id_to_number, is_shown_on_r_all, + newest_consistent_id, number_to_comment_id, ) from racb.core import shutdown @@ -70,6 +73,19 @@ def make_reader(head=10_000, saved=None, fetch_block=None, handle_comment=None, return reader, store +class FakeClock: + """Stands in for time.time(), so tests decide how much time passes between head polls.""" + + def __init__(self, start=1_800_000_000.0): + self.now = start + + def __call__(self): + return self.now + + def advance(self, seconds): + self.now += seconds + + def queued_blocks(reader): blocks = [] while not reader.blocks.empty(): @@ -264,9 +280,13 @@ def test_the_reader_skips_ahead_when_too_far_behind(): head = {'value': 10_000} reader, _ = make_reader(metrics=metrics) reader.fetch_head = lambda: head['value'] - reader.poll_head() - head['value'] = 10_000 + MAX_BACKLOG_IDS + 500 - reader.poll_head() + clock = FakeClock() + with patch('racb.comment_reader.time.time', clock): + reader.poll_head() + # A real backlog that large only builds up over time, e.g. while the head polls failed + clock.advance(600) + head['value'] = 10_000 + MAX_BACKLOG_IDS + 500 + reader.poll_head() metrics.record_skipped_ids.assert_called_once_with(MAX_BACKLOG_IDS + 500) assert reader.next_id == head['value'] + 1 @@ -437,3 +457,203 @@ def test_a_failing_cursor_save_is_logged_not_raised(): store.save_stream_cursor = MagicMock(side_effect=RuntimeError('db down')) with patch.dict('os.environ', {'DEBUG': 'false'}): reader.save_cursor() + + +# Regression tests for the 2026-10-17 outage. The r/all listing returned a comment whose ID was about 6 * 10^18 +# (real IDs were about 8 * 10^10). The reader skipped ahead to it, no real head was ever above its position +# again, and it read nothing for hours while the heartbeat said `reads: 0 requests` and a huge negative lag. +BOGUS_HEAD = 6_022_465_357_511_254_311 +REAL_HEAD = comment_id_to_number('ph9dx3x') + + +@pytest.mark.unit +def test_newest_consistent_id_ignores_an_outlier_in_the_listing(caplog): + import logging + ids = [number_to_comment_id(REAL_HEAD - offset) for offset in range(99)] + ids.append(number_to_comment_id(BOGUS_HEAD)) + with caplog.at_level(logging.WARNING, logger='racb.comment_reader'): + assert newest_consistent_id(ids) == REAL_HEAD + assert [record.levelname for record in caplog.records] == ['WARNING'] + + +@pytest.mark.unit +def test_newest_consistent_id_skips_malformed_ids(): + assert newest_consistent_id(['ph9dx3x', '', 'not-base-36!', None]) == REAL_HEAD + assert newest_consistent_id(['!!']) is None + assert newest_consistent_id([]) is None + + +@pytest.mark.unit +def test_fetch_head_ignores_a_bogus_id_in_the_listing(): + listing = [SimpleNamespace(id=number_to_comment_id(REAL_HEAD - offset)) for offset in range(50)] + listing.insert(10, SimpleNamespace(id=number_to_comment_id(BOGUS_HEAD))) + instance = MagicMock() + instance.subreddit.return_value.comments.return_value = iter(listing) + with patch('racb.comment_reader.reddit.get_reddit_instance', return_value=instance): + assert comment_reader.fetch_head() == REAL_HEAD + + +@pytest.mark.unit +def test_a_bogus_head_far_ahead_is_rejected_not_followed(caplog): + import logging + metrics = MagicMock() + head = {'value': REAL_HEAD} + reader, _ = make_reader(metrics=metrics) + reader.fetch_head = lambda: head['value'] + clock = FakeClock() + with patch('racb.comment_reader.time.time', clock), \ + caplog.at_level(logging.WARNING, logger='racb.comment_reader'): + reader.poll_head() + for _ in range(5): + clock.advance(10) + head['value'] = BOGUS_HEAD + reader.poll_head() + assert reader.next_id == REAL_HEAD + 1 + metrics.record_skipped_ids.assert_not_called() + assert metrics.record_rejected_head.call_count == 5 + # One WARNING for the whole streak, not one per poll + assert [record.levelname for record in caplog.records] == ['WARNING'] + + +@pytest.mark.unit +def test_reading_continues_through_bogus_heads(): + # Replays the incident: real heads, then bogus ones mixed in, then real ones again + metrics = MagicMock() + head = {'value': REAL_HEAD} + reader, _ = make_reader(metrics=metrics) + reader.fetch_head = lambda: head['value'] + clock = FakeClock() + real_head = REAL_HEAD + queued = [] + with patch('racb.comment_reader.time.time', clock): + reader.poll_head() + for poll in range(30): + clock.advance(10) + real_head += 1000 # About 100 new comments a second + head['value'] = BOGUS_HEAD if poll % 3 == 1 else real_head + reader.poll_head() + queued.extend(queued_blocks(reader)) + assert len(queued) == (real_head - REAL_HEAD) // BLOCK_SIZE + assert queued == sorted(set(queued)) + assert reader.next_id <= real_head + 1 + lag_values = [call.args[0] for call in metrics.set_lag_ids.call_args_list] + assert all(0 <= lag < MAX_BACKLOG_IDS for lag in lag_values) + metrics.record_skipped_ids.assert_not_called() + + +@pytest.mark.unit +def test_a_head_jump_is_allowed_when_enough_time_passed(): + head = {'value': REAL_HEAD} + reader, _ = make_reader() + reader.fetch_head = lambda: head['value'] + clock = FakeClock() + with patch('racb.comment_reader.time.time', clock): + reader.poll_head() + clock.advance(3600) + jump = MAX_BACKLOG_IDS + 3600 * MAX_COMMENT_IDS_PER_SECOND + head['value'] = REAL_HEAD + jump + reader.poll_head() + assert reader.next_id == REAL_HEAD + jump + 1 + + +@pytest.mark.unit +def test_a_head_slightly_behind_the_last_one_queues_nothing_and_reading_goes_on(): + # Reddit's listing is not strictly ordered over time: a later poll can return an older newest comment + head = {'value': REAL_HEAD} + reader, _ = make_reader() + reader.fetch_head = lambda: head['value'] + reader.poll_head() + head['value'] = REAL_HEAD - 3 * BLOCK_SIZE + reader.poll_head() + assert queued_blocks(reader) == [] + assert reader.next_id == REAL_HEAD + 1 + head['value'] = REAL_HEAD + 2 * BLOCK_SIZE + reader.poll_head() + assert queued_blocks(reader) == [REAL_HEAD + 1, REAL_HEAD + 1 + BLOCK_SIZE] + + +@pytest.mark.unit +def test_a_position_far_ahead_of_every_head_resets_to_the_head(caplog): + # What the deployed reader was stuck in: a fresh start on a bogus first head has no reference to reject it + import logging + head = {'value': BOGUS_HEAD} + reader, _ = make_reader() + reader.fetch_head = lambda: head['value'] + reader.poll_head() + assert reader.next_id == BOGUS_HEAD + 1 + head['value'] = REAL_HEAD + with caplog.at_level(logging.WARNING, logger='racb.comment_reader'): + for _ in range(HEADS_BELOW_POSITION_BEFORE_RESET - 1): + reader.poll_head() + assert reader.next_id == BOGUS_HEAD + 1 + reader.poll_head() + assert reader.next_id == REAL_HEAD + 1 + assert [record.levelname for record in caplog.records] == ['WARNING'] + head['value'] = REAL_HEAD + 2 * BLOCK_SIZE + reader.poll_head() + assert queued_blocks(reader) == [REAL_HEAD + 1, REAL_HEAD + 1 + BLOCK_SIZE] + + +@pytest.mark.unit +def test_a_restart_with_a_bogus_first_head_resumes_from_the_cursor(): + clock = FakeClock() + start = REAL_HEAD - 5 * BLOCK_SIZE + saved_at = datetime.datetime.fromtimestamp(clock() - 60, datetime.timezone.utc).replace(tzinfo=None) + saved = {'next_comment_id': start, 'done_blocks': [start + BLOCK_SIZE], 'updated_at': saved_at} + head = {'value': BOGUS_HEAD} + reader, _ = make_reader(saved=saved) + reader.fetch_head = lambda: head['value'] + with patch('racb.comment_reader.time.time', clock): + reader.poll_head() + assert reader.next_id == start + assert queued_blocks(reader) == [] + clock.advance(10) + head['value'] = REAL_HEAD + reader.poll_head() + assert queued_blocks(reader) == [start, start + 2 * BLOCK_SIZE, start + 3 * BLOCK_SIZE, + start + 4 * BLOCK_SIZE] + + +@pytest.mark.unit +def test_a_restart_with_a_cursor_ahead_of_the_head_starts_at_the_head(): + # The cursor the stalled reader saved on 2026-10-17: just past the bogus ID + saved = {'next_comment_id': BOGUS_HEAD + 1, 'done_blocks': [], + 'updated_at': datetime.datetime(2026, 10, 17, 6, 0)} + reader, _ = make_reader(head=REAL_HEAD, saved=saved) + reader.poll_head() + assert reader.next_id == REAL_HEAD + 1 + assert queued_blocks(reader) == [] + + +@pytest.mark.unit +def test_the_reader_keeps_making_read_requests_while_heads_are_bogus(): + # End to end with reader threads: reading must not stop when bogus heads come in + reads = [] + real_head = {'value': REAL_HEAD} + polls = [] + + def fetch_head(): + polls.append(1) + if len(polls) % 2 == 0: + return BOGUS_HEAD + real_head['value'] += 2 * BLOCK_SIZE + return real_head['value'] + + def fetch_block(bot_name, start): + reads.append(start) + return [fake_comment(start)] + + handled = [] + reader, _ = make_reader(fetch_block=fetch_block, handle_comment=handled.append) + reader.fetch_head = fetch_head + reader.start() + try: + for _ in range(12): + reader.poll_head() + assert wait_until(lambda: len(reads) >= 10) + finally: + shutdown.request_stop() + for thread in reader.threads: + thread.join(WAIT_SECONDS) + assert reader.next_id <= real_head['value'] + 1 + assert len(handled) >= 10 diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 2cbd96f..3d13b43 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -292,6 +292,7 @@ def test_heartbeat_counts_empty_head_polls(caplog): metrics = StreamMetrics() metrics.record_head_poll(1.0) metrics.record_head_poll(3.0, empty=True) + metrics.record_request(1.0, 100) with caplog.at_level(logging.INFO, logger='racb.main'): metrics.log_heartbeat() assert 'head: 2 polls (1 empty), 2.00 s/poll' in caplog.records[-1].getMessage() @@ -320,6 +321,50 @@ def test_heartbeat_reports_reads_and_lag(caplog): assert metrics.interval_failed_blocks == 0 +@pytest.mark.unit +def test_heartbeat_counts_rejected_heads(caplog): + import logging + metrics = StreamMetrics() + metrics.record_head_poll(1.0) + metrics.record_head_poll(1.0, empty=True) + metrics.record_rejected_head() + metrics.record_request(1.0, 100) + with caplog.at_level(logging.INFO, logger='racb.main'): + metrics.log_heartbeat() + assert 'head: 2 polls (1 empty, 1 rejected), 1.00 s/poll' in caplog.records[-1].getMessage() + assert metrics.interval_rejected_heads == 0 + + +@pytest.mark.unit +def test_heartbeat_warns_when_heads_were_found_but_nothing_was_read(caplog): + # The 2026-10-17 outage: head polls kept succeeding, the readers made no request for hours + import logging + metrics = StreamMetrics() + metrics.record_head_poll(1.0) + with caplog.at_level(logging.INFO, logger='racb.main'): + metrics.log_heartbeat() + assert [record.levelname for record in caplog.records] == ['INFO', 'WARNING'] + assert 'stalled' in caplog.records[-1].getMessage() + + +@pytest.mark.unit +@pytest.mark.parametrize('head_polls, empty_head_polls, requests', [ + (0, 0, 0), # STREAM_SOURCE=listing: no head polls + (3, 3, 0), # Only empty listings: no head to read up to + (3, 0, 5), # Reading normally +]) +def test_heartbeat_does_not_warn_without_a_stall(caplog, head_polls, empty_head_polls, requests): + import logging + metrics = StreamMetrics() + for poll in range(head_polls): + metrics.record_head_poll(1.0, empty=poll < empty_head_polls) + for _ in range(requests): + metrics.record_request(1.0, 100) + with caplog.at_level(logging.INFO, logger='racb.main'): + metrics.log_heartbeat() + assert [record.levelname for record in caplog.records] == ['INFO'] + + @pytest.mark.unit def test_main_loop_records_each_batch_as_one_request(make_mock_comment): from racb.main import main_loop, metrics