From 25dd039f4cab5d042b0eaea1fdcf690ffdc96d22 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 20:02:49 +0000 Subject: [PATCH 1/3] feat: read every new comment by ID range, resume after restarts The r/all comment listing returns at most 100 comments per request, one request at a time, so the bot saw about 850 comments a minute by day and missed most of r/all. Comment IDs are numbered in order site-wide, so the bot now fetches blocks of 100 consecutive IDs with /api/info on STREAM_READERS threads spread over the 4 accounts, keeping only what r/all would show. STREAM_SOURCE=listing keeps the old stream as a fallback. The reader saves its position (stream_cursors) every 30 s and at shutdown, so restarts and redeploys resume instead of skipping comments. The listing stream now resumes after the newest comment seen after an error. The heartbeat also reports requests, comments per request, seconds per request, how far behind the newest comment the bot is, and CPU use (process share of a core and VM steal time). Co-Authored-By: Toldry Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FYGhPFNh3ZvVEr2YXt8Qmr --- .env.example | 5 + AGENTS.md | 100 +++-- changelog.d/+heartbeat-reading-stats.added.md | 1 + changelog.d/+read-comments-by-id.added.md | 1 + changelog.d/+resume-comment-reading.added.md | 1 + src/racb/comment_reader.py | 319 ++++++++++++++++ src/racb/config.py | 30 ++ src/racb/core/cpu_usage.py | 56 +++ src/racb/core/db.py | 16 + src/racb/main.py | 173 ++++++++- src/racb/sql/instantiate_db.sql | 42 +++ src/racb/workers.py | 29 +- tests/integration/test_reddit_integration.py | 22 ++ tests/unit/test_comment_reader.py | 350 ++++++++++++++++++ tests/unit/test_config.py | 22 ++ tests/unit/test_cpu_usage.py | 35 ++ tests/unit/test_main.py | 88 +++++ 17 files changed, 1240 insertions(+), 50 deletions(-) create mode 100644 changelog.d/+heartbeat-reading-stats.added.md create mode 100644 changelog.d/+read-comments-by-id.added.md create mode 100644 changelog.d/+resume-comment-reading.added.md create mode 100644 src/racb/comment_reader.py create mode 100644 src/racb/core/cpu_usage.py create mode 100644 tests/unit/test_comment_reader.py create mode 100644 tests/unit/test_cpu_usage.py diff --git a/.env.example b/.env.example index a20cdf5..5494d3f 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,11 @@ PHASE2_WAITING_PERIOD=2 hours PHASE3_WAITING_PERIOD=3 months # Threads that run Phase 1 (duplicate search, auxiliary replies) on matching r/all comments PHASE1_WORKERS=4 +# How new comments are read: 'ids' (every new comment, by ID range) or 'listing' (the r/all listing, which +# misses most comments; kept as a fallback) +STREAM_SOURCE=ids +# With STREAM_SOURCE=ids: threads fetching blocks of 100 comment IDs, taking turns between the 4 accounts +STREAM_READERS=4 # ------------------------------------------------------------------------------ # 1. Primary Bot: AutoCrosspostBot (u/AutoCrosspostBot) diff --git a/AGENTS.md b/AGENTS.md index 66668c0..7fd184c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,9 +12,10 @@ Welcome to the **RedditAutoCrosspostBot** repository. This document is a compreh This repository hosts a suite of 4 automated Reddit bots powered by [PRAW (Python Reddit API Wrapper)](https://praw.readthedocs.io/), backed by a PostgreSQL database and containerized with Docker / Docker Compose for deployment on free cloud VPS instances (e.g. Google Cloud Always Free or Oracle Cloud Always Free) or local hardware. -All 4 bots run in **one Python process, on several threads**. The main thread only reads the `r/all` -comment stream. Phase 1 runs on a pool of worker threads, and every scheduled job (Phase 2, Phase 3, inbox, -cleanup, ...) on a thread of its own, so a slow job never pauses the stream (see section 7). +All 4 bots run in **one Python process, on several threads**. New comments are read by ID on a few +reader threads (`comment_reader.py`, section 7), with the main thread tracking the newest comment ID. +Phase 1 runs on a pool of worker threads, and every scheduled job (Phase 2, Phase 3, inbox, cleanup, ...) +on a thread of its own, so a slow job never pauses reading. ### Primary Bot: `AutoCrosspostBot` - **Goal:** Monitors Reddit comments on `r/all` for community crosspost recommendations matching `r/SomeSubreddit` or `/r/SomeSubreddit`. @@ -105,13 +106,19 @@ A banned target (a 404 whose body says `"reason": "banned"`, `SUBREDDIT_IS_BANNE ### Phase Details - **Phase 1 ([`src/racb/phases/phase1.py`](./src/racb/phases/phase1.py)):** - - Continuous streaming from `reddit.subreddit('all').stream.comments(skip_existing=True, pause_after=-1)`. + - Continuous reading of every new comment by ID (`comment_reader.CommentReader`, the default, + `STREAM_SOURCE=ids`): blocks of 100 consecutive comment IDs fetched with `reddit.info()` on + `STREAM_READERS` threads, keeping only what r/all would show (`is_shown_on_r_all`: not NSFW, quarantined, + private or on a user profile). Subreddits that opted out of r/all cannot be told apart, so their comments + are read too. `STREAM_SOURCE=listing` instead reads + `reddit.subreddit('all').stream.comments(skip_existing=True, pause_after=-1)` one request at a time, which + sees only a fraction of r/all (section 9). - Filters: Discards moderator comments, submissions with meta-phrases (`sub`, `subreddit`), and blacklisted subreddits ([`src/racb/constants.py`](./src/racb/constants.py), compared case-insensitively by `is_crosspost_blocked`). Non-top-level comments (`comment.parent_id != comment.link_id`) are discarded only **after** auxiliary-bot dispatch (see the warning in section 1). - Qualified suggestions are stored via stored procedure `insert_scraped_comment(permalink, target_subreddit)`. - Phases 2 and 3 re-parse the target from the live comment body and reject the row (`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 stream thread only runs the regex (`check_pattern`, no I/O) and queues the comment ID of a match. A + - 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 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. @@ -201,7 +208,9 @@ A banned target (a 404 whose body says `"reason": "banned"`, `SUBREDDIT_IS_BANNE | File | Purpose | | :--- | :--- | -| [`src/racb/main.py`](./src/racb/main.py) | Application entry point, logging setup, the stream loop on the main thread, exception recovery, the job list (`build_jobs`), and graceful shutdown on SIGTERM/SIGINT. | +| [`src/racb/main.py`](./src/racb/main.py) | Application entry point, logging setup, `StreamMetrics` (the heartbeat), `run_stream` (starts the comment reader, or the listing stream loop with `STREAM_SOURCE=listing`), exception recovery, the job list (`build_jobs`), and graceful shutdown on SIGTERM/SIGINT. | +| [`src/racb/comment_reader.py`](./src/racb/comment_reader.py) | `CommentReader`: reads every new comment by blocks of 100 consecutive IDs on `STREAM_READERS` threads spread over the 4 accounts, and saves its position (`stream_cursors`) so restarts resume. Its module docstring explains the method. | +| [`src/racb/core/cpu_usage.py`](./src/racb/core/cpu_usage.py) | Process CPU time and VM steal time (`/proc/stat`) between heartbeats. | | [`src/racb/workers.py`](./src/racb/workers.py) | Background threads: `PeriodicJob` (one thread per scheduled job), `Phase1Workers` (the Phase 1 thread pool fed by the stream), and `run_job_safely`. | | [`src/racb/__main__.py`](./src/racb/__main__.py) | Module executable entrypoint for `python -m racb`. | | [`src/racb/__init__.py`](./src/racb/__init__.py) | Package marker; re-exports `__version__`. | @@ -286,6 +295,8 @@ Store these in `.env` for local development or production: | `PHASE2_WAITING_PERIOD` | Time string | `2 hours` | Time duration before Phase 2 checks a comment (parsed via `pytimeparse`, e.g. `2 hours` or `1 day`). | | `PHASE3_WAITING_PERIOD` | Time string | `90 days` | Time duration before Phase 3 executes the crosspost (e.g. `3 months` or `90 days`). The public crosspost reply shows it as a localized whole number of days. | | `PHASE1_WORKERS` | Integer | `4` | Number of Phase 1 worker threads (at least 1). Each makes its own Reddit requests, all sharing the accounts' rate limits (section 7). | +| `STREAM_SOURCE` | `ids` / `listing` | `ids` | How new comments are read: `ids` reads every comment by ID range (`comment_reader.py`), `listing` reads PRAW's r/all comment stream one request at a time (the pre-3.1 way, kept as a fallback). An unknown value logs a WARNING and uses `ids`. | +| `STREAM_READERS` | Integer | `4` | With `STREAM_SOURCE=ids`: threads fetching ID blocks (at least 1), taking turns between the 4 accounts. Raising it only helps if the heartbeat's `behind:` keeps growing; requests per minute are set by how fast Reddit gets new comments, not by this. | | `LOG_DIR` | Path | `logs` | Directory for `app.log` (relative to the CWD; `/app/logs` in Docker, mounted on volume `racb_logs`). | | `PASSWORD` | String | | Account password for `AutoCrosspostBot`. | | `APP_CLIENT_SECRET` | String | | App client secret for `AutoCrosspostBot`. | @@ -424,8 +435,9 @@ uv run pytest -m integration ### Logging - Format: `YYYY-MM-DD HH:MM:SS [LEVEL ] [thread] module.name: message`, sent to stderr and to - `$LOG_DIR/app.log` (rotating, 5 x 5 MB). `thread` is `MainThread` (the stream), `phase1-`, or the - job's name (`phase2`, `phase3`, `inbox`, `cleanup`, `heartbeat`, `known-subreddits`). + `$LOG_DIR/app.log` (rotating, 5 x 5 MB). `thread` is `MainThread` (head polls, or the listing stream), + `reader-` (comment ID blocks), `phase1-`, or the job's name (`phase2`, `phase3`, `inbox`, `cleanup`, + `heartbeat`, `known-subreddits`). - The `prawcore` and `urllib3.connectionpool` loggers are **disabled entirely**, which also hides prawcore's rate-limit messages. - `[DRY RUN]` lines show writes that `DRY_RUN` suppressed. `Skipping comment ... after Reddit error` lines show per-comment failures (e.g. a bot banned from a subreddit) that no longer interrupt the stream. @@ -433,8 +445,16 @@ uv run pytest -m integration (Phase 1 auxiliary replies and the crosspost reply) or `/u/ replied to feedback (r/): ` (inbox). A Phase 1 `Replying via ...` line without a matching `replied to` line means the reply failed. - A heartbeat line (`[Heartbeat] Last ~10m: scanned N comments ...`) is logged every 10 minutes by `StreamMetrics`. It is the quickest health and throughput signal. - It ends with `N match(es) dropped, Phase 1 queue full` when the Phase 1 workers fell behind. Besides that, + It says `N match(es) dropped, Phase 1 queue full` when the Phase 1 workers fell behind. Besides that, only two lines are logged per backlog: when the queue fills and when it has room again. + Its second half says how comments were read and what it cost, for example + `reads: 412 requests, 87.3 comments/request, 0.95 s/request | head: 160 polls, 1.70 s/poll | behind: 9 s, 340 IDs | CPU: 4.1% of a core, steal 0.6%`: + - `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. + - `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. + - `CPU`: the process's CPU time as a share of one core, and the VM's steal time (section 8). ### Release & Semantic Versioning Workflow We follow [Semantic Versioning (SemVer)](https://semver.org/) and [Keep a Changelog](https://keepachangelog.com/). **Releases are fully automatic; nobody bumps the version by hand.** @@ -525,6 +545,8 @@ Defined in [`src/racb/sql/instantiate_db.sql`](./src/racb/sql/instantiate_db.sql | `known_subreddits` | `name_lower VARCHAR(32) PK`, `display_name VARCHAR(32) NOT NULL`, `last_seen TIMESTAMP NOT NULL`. GIN trigram index `idx_known_subreddits_name_trgm` on `name_lower`. Every subreddit (except `u_` profiles) the stream has seen a comment in; rows are never deleted, so a later-banned subreddit stays (its suggestion line then says so). | | `record_known_subreddits(text[])` | Upserts a batch of names, refreshing `display_name` and `last_seen`. | | `find_similar_subreddits(varchar, int)` | `(display_name, similarity)` for names matching `name_lower % lower(query)` (trigram similarity ≥ `pg_trgm.similarity_threshold`, default 0.3), most similar first. | +| `stream_cursors` | `name VARCHAR(32) PK`, `next_comment_id BIGINT NOT NULL`, `done_blocks BIGINT[] NOT NULL`, `updated_at TIMESTAMP NOT NULL`. Where `comment_reader` stopped (row `comments`): every comment ID (base 36 as a number) below `next_comment_id` was read, and so were the 100-ID blocks starting at each of `done_blocks`. | +| `save_stream_cursor(varchar, bigint, bigint[])` / `get_stream_cursor(varchar)` | Upsert / read one cursor; `get` returns no row if none is saved. | Behavior to be aware of: - The script runs `CREATE EXTENSION IF NOT EXISTS pg_trgm` inside a `DO` block that only warns on failure. @@ -553,7 +575,10 @@ Behavior to be aware of: ```mermaid flowchart LR - S["MainThread\nr/all stream"] -->|"comment ID of a\nregex match"| Q[["Phase 1 queue\n(1000 IDs)"]] + M["MainThread\npolls the newest comment ID"] -->|"blocks of\n100 IDs"| BQ[["block queue"]] + M -.->|"every 30 s\nand at shutdown"| C[("stream_cursors")] + BQ --> S["reader-1 ... reader-N\n(STREAM_READERS,\n4 accounts in turn)"] + S -->|"comment ID of a\nregex match"| Q[["Phase 1 queue\n(1000 IDs)"]] Q --> W["phase1-1 ... phase1-N\n(PHASE1_WORKERS)"] S -->|"subreddit name\n(no I/O)"| K["known_subreddits buffer"] J1["phase2"] -.-|"one lock"| J2["phase3"] @@ -578,11 +603,28 @@ share a lock and never run at the same time, because both take rows from `scrape re-runs Phase 2's filters. Every run is wrapped in `workers.run_job_safely()`, which logs a failing job's exception instead of ending its thread, and calls `db.reset_connection()` on DB connection errors. -`main_loop()` reads one batch from the 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. For each comment, the main 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 counted in the heartbeat, so the stream never waits for the workers. +**Reading comments (`STREAM_SOURCE=ids`, `comment_reader.CommentReader`).** Reddit numbers comments in +base 36, in order across the whole site. The main thread asks the r/all listing for its newest comment +(one request, then a 10 s pause, which only delays reading, never skips anything) and queues every complete block of 100 IDs up to it. Each `reader-` +thread takes a block, fetches it with `reddit.info(fullnames=...)` (one request, as its account) and passes +each comment r/all would show to `main.dispatch_comment`, several threads at once. So requests per minute +follow Reddit's comment rate (about one per 100 new comments), whatever `STREAM_READERS` is. A block that +fails with a network error is retried after the usual back-off, up to 3 tries, then given up on and counted. + +The reader saves its position every 30 s and at shutdown: the lowest ID not read yet (the watermark), and +the blocks already done above it, so a restart neither skips nor re-reads them. A restart more than +`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). + +**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 +error, the new stream continues after the newest comment seen (`continue_after_id`); after a restart it +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 +counted in the heartbeat, so reading never waits for the workers. Rules for thread safety: @@ -594,10 +636,11 @@ Rules for thread safety: stay close to the quota and all wait for the reset once it is used up. They can still overshoot by a few requests at once, which ends in a `TooManyRequests` that the thread backs off from (60 s). - **Each thread has its own database connection** (`db.get_connection()`, thread-local). -- Module-level state written by more than one thread needs a lock (`StreamMetrics`, `known_subreddits`). +- Module-level state written by more than one thread needs a lock (`StreamMetrics`, `known_subreddits`, + `Phase1Workers.submit`, `CommentReader`'s block bookkeeping). -**Shutdown:** SIGTERM (`docker stop`) and SIGINT set the stop signal in `racb.core.shutdown`. The stream -loop, the Phase 1 workers and the job threads check it, every back-off waits with `shutdown.stop_requested_within()` +**Shutdown:** SIGTERM (`docker stop`) and SIGINT set the stop signal in `racb.core.shutdown`. The main +thread, the reader threads, the Phase 1 workers and the job threads check it, every back-off waits with `shutdown.stop_requested_within()` instead of `time.sleep()`, and Phase 2 and Phase 3 stop between entries, so an entry is never left crossposted but not deleted. `start_bot()` waits up to `SHUTDOWN_TIMEOUT_SECONDS` (50 s) for all threads; `docker-compose.yml` gives the bot `stop_grace_period: 60s`. Matches still waiting in the Phase 1 queue are @@ -614,22 +657,24 @@ flowchart TD S -->|"DB connection error"| R["reset DB connection,\nskip comment"] S -->|"other Reddit / PRAW error\n(e.g. 403 bot banned)"| K["log INFO, skip comment"] S -->|"anything else"| L["log with traceback, skip comment"] - F["stream fetch\non the main thread"] -->|raises| H{"handle_exception()"} + F["head poll or listing fetch\non the main thread"] -->|raises| H{"handle_exception()"} H -->|"TooManyRequests"| W60["sleep 60 s"] H -->|"network error or Forbidden"| W30["sleep 30 s"] H -->|"DB connection error"| WDB["reset DB, sleep 30 s"] H -->|"graceful RedditAPIException"| N["no sleep"] - W60 & W30 & WDB & N --> RS["re-create stream\n(skip_existing=True)"] + W60 & W30 & WDB & N --> RS["poll again (ids), or re-create\nthe stream after the newest\ncomment seen (listing)"] H -->|"anything else"| X["crash; Docker restarts"] ``` - Matching uses `isinstance`, so subclasses are covered. - With `DEBUG=True`, every layer re-raises. -- A stream reset uses `skip_existing=True`, so comments posted during the outage are never seen. +- A failing head poll only stops new blocks from being queued; the readers finish the ones queued and the + next poll catches up, so nothing is lost. Reader threads back off from their own errors + (`workers.back_off_after_network_error`) and retry the block. Rule for new code (reinforcing rule 3 in section 10): catch per-item errors **inside** the per-comment or per-row handler. Only network-level failures should reach the worker's back-off or `handle_exception()`. -A stream back-off only pauses reading `r/all`; the workers and jobs keep running. +A back-off on the main thread only pauses finding new comments; the readers, workers and jobs keep running. --- @@ -650,6 +695,13 @@ A stream back-off only pauses reading `r/all`; the workers and jobs keep running the image contains `.env`. They `git reset --hard` the VM checkout, so `scripts/redeploy.sh` refuses to run when tracked files were edited on the VM. Keep `redeploy.sh`'s body inside `main()`: the reset rewrites the running script. +- **CPU on the Always Free e2-micro:** the free tier covers the instance itself, not a CPU budget, so CPU use + is not billed; the VM gets about a quarter of a core over time and is throttled beyond that. The heartbeat's + `CPU:` part shows both sides: the process's share of one core (sustained values near 25% mean the bot uses + what the VM gets) and steal (the hypervisor holding the VM back). What the free tier does limit is network + egress, 1 GB a month (https://cloud.google.com/free/docs/free-cloud-features#compute); downloads from Reddit + are ingress, but every request and its TCP acknowledgements go out. Check the VM's network egress in Cloud + Monitoring after a change that adds requests. - Python versions: `requires-python = ">=3.11"`. The Docker image, CI and `.python-version` use 3.11, and the unit tests also pass on 3.13. 3.10 is not supported. - Mermaid diagrams: GitHub's renderer rejects unquoted labels that contain parentheses, e.g. @@ -664,7 +716,9 @@ A stream back-off only pauses reading `r/all`; the workers and jobs keep running | Duplicate crossposts | Before every crosspost, Phase 3 re-runs `get_posts_with_same_content()`. That is a Reddit search for the submission URL in the target (subject to search-index lag of minutes) plus Reddit's "other discussions" of the source post, a search for the same title, and a scan of every image (or video) post the target got since the source comment was written (a live listing, so it catches a crosspost made seconds earlier, but only for **image (including gallery) and video** posts, galleries by their first 10 pictures, crossposts through their original post, videos only by preview frame, and in busy subreddits only as far back as Reddit's ~1000-post listing reaches). The remaining gap is link and text posts whose title was changed and that are not linked as "other discussions", when two rows for the same submission and target are handled in one Phase 3 run, or after a crash between `crosspost()` and the row delete. The owner considers this acceptable, and Reddit may also reject duplicates server-side (unverified). An `ALREADY_SUB`-style rejection is not in `familiar_error_types`, so it would be logged as unexpected. If the reply fails after a successful crosspost, the crosspost is reported as failed. | | Negative-score cleanup | Deletes `AutoCrosspostBot` submissions with `score < 0`. The owner reports that submission scores can go negative; this was not independently confirmed. | | Phase 1 backlog | Phase 1 (duplicate search, up to 200 preview downloads per match, subreddit suggestions) runs on `PHASE1_WORKERS` threads. If matches arrive faster than they can handle for long enough to fill the 1000-ID queue, new matches are dropped; the heartbeat counts them. Not observed; matches are a small fraction of `r/all`. | -| Stream outages | A stream reset after a network error still uses `skip_existing=True`, so comments posted while the stream was down are never seen. Jobs never pause the stream, but PRAW's stream only reads the newest 100 comments per request, so if `r/all` outpaces the request rate, comments are skipped (unverified how often). | +| Comment coverage | `STREAM_SOURCE=listing` reads at most 100 comments per request, one request at a time. On 2026-10-15/16 that was about 850 comments a minute by day and up to 2,500 at night, when Reddit answered faster, so it missed most of r/all. The `ids` reader was written for that; it relies on Reddit numbering comments in order site-wide, which `test_live_comment_reader_reads_recent_comments_by_id` checks before every deploy, and its real coverage is still to be confirmed from the heartbeat. | +| Re-reads after a crash | The comment cursor is saved every 30 s and at shutdown. After a crash or `kill -9`, up to 30 s of comments are read again, so an auxiliary bot could reply twice to a comment from those seconds. | +| Opted-out subreddits | `/api/info` does not say whether a subreddit opted out of r/all, so the `ids` reader also reads comments in subreddits that did (NSFW, quarantined, private and profile comments are dropped). | | Rotated reposts | Not detected as duplicates (see the image-matching note in section 5). | | Edited video reposts | Videos are compared by their preview frame only. A trimmed, re-cut or re-encoded copy usually has a different preview frame and is missed. | | Pre-v2.3 rows | Have `target_subreddit = NULL`, so the bait-and-switch check is skipped for them until they age out (at most one Phase 3 waiting period after upgrading). | diff --git a/changelog.d/+heartbeat-reading-stats.added.md b/changelog.d/+heartbeat-reading-stats.added.md new file mode 100644 index 0000000..22286c0 --- /dev/null +++ b/changelog.d/+heartbeat-reading-stats.added.md @@ -0,0 +1 @@ +**Reading and CPU Figures in the Heartbeat:** each heartbeat now also says how many requests read comments, how many comments each returned and how long each took, how far behind the newest comment the bot is, and its CPU use (share of one core, and the VM's steal time), so you can tell whether it keeps up with Reddit and whether the VM is the limit. diff --git a/changelog.d/+read-comments-by-id.added.md b/changelog.d/+read-comments-by-id.added.md new file mode 100644 index 0000000..6ddc237 --- /dev/null +++ b/changelog.d/+read-comments-by-id.added.md @@ -0,0 +1 @@ +**Every New Comment, Read by ID:** the bot read r/all through its comment listing, 100 comments per request and one request at a time, which saw about 850 comments a minute by day and missed most of r/all. It now reads every new comment by blocks of 100 consecutive comment IDs on `STREAM_READERS` threads (default 4) spread over the 4 bot accounts, keeping only what r/all would show. `STREAM_SOURCE=listing` restores the old way. diff --git a/changelog.d/+resume-comment-reading.added.md b/changelog.d/+resume-comment-reading.added.md new file mode 100644 index 0000000..fa5b5e7 --- /dev/null +++ b/changelog.d/+resume-comment-reading.added.md @@ -0,0 +1 @@ +**No Comments Skipped on Restarts:** the comment reader saves its position in the database (`stream_cursors`) every 30 seconds and at shutdown, so a restart or redeploy resumes where it stopped instead of skipping what was posted meanwhile (up to about 200,000 comment IDs back). With `STREAM_SOURCE=listing`, a stream error resumes after the newest comment seen instead of skipping ahead. diff --git a/src/racb/comment_reader.py b/src/racb/comment_reader.py new file mode 100644 index 0000000..0149b59 --- /dev/null +++ b/src/racb/comment_reader.py @@ -0,0 +1,319 @@ +"""Reads every new Reddit comment by ID range, instead of polling the r/all comment listing. + +Why: the r/all comment listing returns at most 100 comments per request, and each request takes several +seconds. On 2026-10-15/16 the bot read about 850 comments a minute through it during the US day and up +to 2,500 a minute at night, when Reddit answers faster. So the listing, not r/all, set the count, and most +comments were never seen. + +Comment IDs are base-36 numbers that Reddit hands out in order across the whole site. So instead the bot +asks for 100 consecutive IDs at a time (`/api/info`), on several threads and accounts at once, without +gaps or overlap: + +- The main thread asks the r/all listing for its newest comment (the "head") and queues the blocks of + 100 IDs up to it. +- `STREAM_READERS` threads, spread over the 4 bot accounts, each fetch one block at a time and pass every + comment that r/all would show (`is_shown_on_r_all`) to `handle_comment`. +- The ID below which every block is done (the "watermark") and the blocks already done above it are + saved to the database every `CURSOR_SAVE_SECONDS` and at shutdown, so a restart resumes where the previous + run stopped instead of skipping what was posted meanwhile. + +`/api/info` also returns comments that r/all leaves out (private and NSFW subreddits, user profiles), so +those are dropped. Subreddits that only opted out of r/all cannot be told apart from a comment, so their +comments are now read too. +""" + +import logging +import queue +import threading +import time + +from racb import config +from racb.core import db, reddit, shutdown +from racb.workers import POLL_SECONDS, back_off_after_network_error + +logger = logging.getLogger(__name__) + +# /api/info returns at most 100 items per request +BLOCK_SIZE = 100 + +# Seconds to wait between two head polls (each poll itself takes a few seconds). It only sets how soon a new +# comment is read, not how many: every block up to the head is queued at once. Fewer polls, fewer requests. +HEAD_POLL_SECONDS = 10 + +# At most this many IDs may wait to be read (about half an hour of Reddit comments, at a guessed 100 per +# second). Further behind, the reader skips ahead to the head and logs how many IDs it skipped, so it never +# works on comments too old to be worth a reply. The same limit decides whether a restart resumes. +MAX_BACKLOG_IDS = 200_000 + +# Tries per block before its comments are given up on +MAX_BLOCK_ATTEMPTS = 3 + +CURSOR_SAVE_SECONDS = 30 +CURSOR_NAME = 'comments' + +# Subreddit types whose comments never appear on r/all +HIDDEN_SUBREDDIT_TYPES = frozenset({'private', 'user', 'employees_only', 'gold_only'}) + +BASE36_DIGITS = '0123456789abcdefghijklmnopqrstuvwxyz' + + +def comment_id_to_number(comment_id): + """`'k3l2x9a'` -> its number. Also accepts a fullname (`'t1_k3l2x9a'`).""" + if comment_id.startswith('t1_'): + comment_id = comment_id[3:] + return int(comment_id, 36) + + +def number_to_comment_id(number): + """The inverse of `comment_id_to_number`, without the `t1_` prefix.""" + if number == 0: + return '0' + digits = [] + while number: + number, remainder = divmod(number, 36) + digits.append(BASE36_DIGITS[remainder]) + return ''.join(reversed(digits)) + + +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)] + + +def is_shown_on_r_all(comment): + """Whether r/all would list this comment: not NSFW, quarantined, private or on a user profile.""" + # vars(), not getattr: reading a field the comment lacks would make PRAW fetch it, one request each + fields = vars(comment) + if fields.get('over_18') or fields.get('quarantine'): + return False + if fields.get('subreddit_type') in HIDDEN_SUBREDDIT_TYPES: + return False + subreddit = fields.get('subreddit') + subreddit_name = getattr(subreddit, 'display_name', subreddit) or '' + return not str(subreddit_name).lower().startswith('u_') + + +def fetch_head(): + """Returns the number of the newest comment on r/all (one request, as /u/AutoCrosspostBot).""" + newest = list(reddit.get_reddit_instance().subreddit('all').comments(limit=1)) + if not newest: + raise RuntimeError('The r/all comment listing returned no comments') + return comment_id_to_number(newest[0].id) + + +def fetch_block(bot_name, block_start): + """Returns the comments of one block that still exist and are visible (one request, as `bot_name`).""" + return list(reddit.get_reddit_instance(bot_name).info(fullnames=block_fullnames(block_start))) + + +class CommentReader: + """Reads every new comment by ID block on reader threads; see the module docstring. + + `handle_comment(comment)` runs on the reader threads, several at once, so it must be thread safe. + `run()` occupies the calling thread (the main thread) until shutdown. + """ + + def __init__(self, handle_comment, reader_count, metrics=None, handle_exception=None, + fetch_head=fetch_head, fetch_block=fetch_block, cursor_store=db): + self.handle_comment = handle_comment + self.metrics = metrics + # Decides what to do about an error from a head poll: returns True to re-raise (see main.handle_exception) + self.handle_exception = handle_exception or (lambda error: True) + self.fetch_head = fetch_head + self.fetch_block = fetch_block + self.cursor_store = cursor_store + self.blocks = queue.Queue() + self._lock = threading.Lock() + # The next ID to queue; None until the start position is known + self.next_id = None + # Starts of the blocks queued but not done yet, and of the blocks done above the watermark + self.in_flight = set() + self.done_above = set() + self.last_saved = time.monotonic() + # Readers take turns between the accounts, so each account's rate limit carries a share + self.threads = [ + threading.Thread( + target=self.read_blocks, + args=(reddit.BOT_NAMES[number % len(reddit.BOT_NAMES)],), + name=f'reader-{number + 1}', + daemon=True, + ) + for number in range(reader_count) + ] + + def start(self): + for thread in self.threads: + thread.start() + return self + + # --- Main thread --- + + def run(self): + """Polls the head and queues blocks until shutdown, then saves the cursor.""" + try: + while not shutdown.stop_requested(): + try: + self.poll_head() + except Exception as error: + if self.handle_exception(error): + raise + self.save_cursor_if_due() + if shutdown.stop_requested_within(HEAD_POLL_SECONDS): + break + finally: + self.save_cursor() + + def poll_head(self): + started = time.monotonic() + head = self.fetch_head() + if self.metrics: + self.metrics.record_head_poll(time.monotonic() - started) + with self._lock: + if self.next_id is None: + self._choose_start(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 _choose_start(self, head): + """Resumes from the saved cursor if it is recent enough, otherwise starts at the head.""" + saved = self._load_cursor() + 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} + logger.info( + f'Resuming the comment reader {head + 1 - self.next_id:,} IDs behind the newest comment ' + f'(cursor saved at {saved["updated_at"]})' + ) + return + if saved is not None: + logger.warning( + f'Saved comment cursor is {head + 1 - saved["next_comment_id"]:,} IDs from the newest comment ' + f'(more than {MAX_BACKLOG_IDS:,}, or ahead of it); starting at the newest comment instead' + ) + self.next_id = head + 1 + logger.info(f'Comment reader starts at the newest comment, ID {number_to_comment_id(head)}') + + def _load_cursor(self): + try: + return self.cursor_store.get_stream_cursor(CURSOR_NAME) + except Exception as error: + if config.is_debug(): + raise + logger.warning(f'Could not read the saved comment cursor ({error!r}); starting at the newest comment') + return None + + def _skip_ahead_if_too_far_behind(self, head): + behind = head + 1 - self.next_id + if behind <= MAX_BACKLOG_IDS: + return + logger.warning(f'Comment reader is {behind:,} IDs behind the newest comment; skipping them') + if self.metrics: + self.metrics.record_skipped_ids(behind) + self.next_id = head + 1 + self.done_above.clear() + + def _queue_blocks_up_to(self, head): + """Queues every complete block up to `head`, as long as fewer than MAX_BACKLOG_IDS wait to be read.""" + # Only complete blocks: the newest IDs of a partial block may not exist yet + while self.next_id + BLOCK_SIZE - 1 <= head and len(self.in_flight) * BLOCK_SIZE < MAX_BACKLOG_IDS: + start = self.next_id + self.next_id += BLOCK_SIZE + if start in self.done_above: + continue + self.in_flight.add(start) + self.blocks.put(start) + + def _watermark(self): + """Every ID below this has been read. Call with the lock held.""" + return min(self.in_flight) if self.in_flight else self.next_id + + def save_cursor_if_due(self): + if time.monotonic() - self.last_saved >= CURSOR_SAVE_SECONDS: + self.save_cursor() + + def save_cursor(self): + """Saves the watermark and the blocks done above it. Errors are logged; the next save retries.""" + self.last_saved = time.monotonic() + with self._lock: + if self.next_id is None: + return + watermark = self._watermark() + self.done_above = {start for start in self.done_above if start >= watermark} + done_blocks = sorted(self.done_above) + try: + self.cursor_store.save_stream_cursor(CURSOR_NAME, watermark, done_blocks) + except Exception as error: + if config.is_debug(): + raise + if isinstance(error, db.CONNECTION_ERRORS): + db.reset_connection() + logger.warning(f'Could not save the comment cursor: {error!r}') + + # --- Reader threads --- + + def read_blocks(self, bot_name): + try: + while not shutdown.stop_requested(): + try: + block_start = self.blocks.get(timeout=POLL_SECONDS) + except queue.Empty: + continue + self.read_block(bot_name, block_start) + except BaseException as error: + shutdown.fail(error, threading.current_thread().name) + + def read_block(self, bot_name, block_start): + """Fetches one block, retrying network errors, and hands its comments on.""" + comments = None + for attempt in range(1, MAX_BLOCK_ATTEMPTS + 1): + started = time.monotonic() + try: + comments = self.fetch_block(bot_name, block_start) + break + except reddit.TRANSIENT_NETWORK_ERRORS as error: + if config.is_debug(): + raise + # A block left unfinished at shutdown is read again after the restart + if back_off_after_network_error(error): + return + if attempt == MAX_BLOCK_ATTEMPTS: + logger.warning(f'Giving up on comment IDs {self._describe(block_start)} after {attempt} tries') + except Exception as error: + if config.is_debug(): + raise + logger.exception(f'Could not read comment IDs {self._describe(block_start)}: {error!r}') + break + if comments is None: + if self.metrics: + self.metrics.record_failed_block() + else: + if self.metrics: + self.metrics.record_request(time.monotonic() - started, len(comments)) + newest_created = None + for comment in comments: + if is_shown_on_r_all(comment): + self.handle_comment(comment) + created = vars(comment).get('created_utc') + if created is not None and (newest_created is None or created > newest_created): + newest_created = created + if self.metrics and newest_created is not None: + self.metrics.record_newest_created(newest_created) + self.finish(block_start) + + def finish(self, block_start): + with self._lock: + self.in_flight.discard(block_start) + if block_start >= self._watermark(): + self.done_above.add(block_start) + + def pending(self): + """Number of blocks queued or being read.""" + with self._lock: + return len(self.in_flight) + + @staticmethod + def _describe(block_start): + return f'{number_to_comment_id(block_start)}-{number_to_comment_id(block_start + BLOCK_SIZE - 1)}' diff --git a/src/racb/config.py b/src/racb/config.py index eebf766..4755d14 100644 --- a/src/racb/config.py +++ b/src/racb/config.py @@ -118,3 +118,33 @@ def phase1_worker_count(): logger.warning(f'Could not parse PHASE1_WORKERS={raw_value!r} as a number; using {PHASE1_WORKERS_DEFAULT}') return PHASE1_WORKERS_DEFAULT return max(1, count) + + +# Where the bot reads new comments from (see racb.comment_reader): +# - 'ids': every new comment on Reddit, by ID range, on STREAM_READERS threads (the default) +# - 'listing': PRAW's stream of the r/all comment listing, one request at a time (the pre-3.1 way) +STREAM_SOURCES = ('ids', 'listing') +STREAM_SOURCE_DEFAULT = 'ids' + + +def stream_source(): + """Returns 'ids' or 'listing', from STREAM_SOURCE.""" + value = os.environ.get('STREAM_SOURCE', STREAM_SOURCE_DEFAULT).strip().lower() + if value not in STREAM_SOURCES: + logger.warning(f'Unknown STREAM_SOURCE={value!r}; using {STREAM_SOURCE_DEFAULT!r}') + return STREAM_SOURCE_DEFAULT + return value + + +STREAM_READERS_DEFAULT = 4 + + +def stream_reader_count(): + """Number of threads that fetch comment ID blocks when STREAM_SOURCE is 'ids' (at least 1).""" + raw_value = os.environ.get('STREAM_READERS', str(STREAM_READERS_DEFAULT)) + try: + count = int(raw_value) + except ValueError: + logger.warning(f'Could not parse STREAM_READERS={raw_value!r} as a number; using {STREAM_READERS_DEFAULT}') + return STREAM_READERS_DEFAULT + return max(1, count) diff --git a/src/racb/core/cpu_usage.py b/src/racb/core/cpu_usage.py new file mode 100644 index 0000000..f8566a9 --- /dev/null +++ b/src/racb/core/cpu_usage.py @@ -0,0 +1,56 @@ +"""Measures the bot's CPU use for the heartbeat, so a CPU-starved VM shows up in the log. + +Two numbers per heartbeat interval: +- The process's CPU time (all threads) as a percentage of one core. On the Always Free e2-micro, the + VM is entitled to about a quarter of a core over time, so sustained values near 25% mean the bot is + using up what the VM gets. +- Steal: the share of the VM's CPU time the hypervisor gave to other machines while the VM wanted to run + (`/proc/stat`). High steal means the VM is being throttled, whatever the bot itself uses. In a Docker + container, `/proc/stat` shows the whole VM, which is what matters here. Elsewhere (macOS, Windows) it is + not available, and only the process's own share is reported. +""" + +import time + +PROC_STAT_PATH = '/proc/stat' + +# Field positions on the `cpu` line of /proc/stat: user nice system idle iowait irq softirq steal ... +STEAL_FIELD_INDEX = 7 + + +def read_steal_and_total_ticks(path=PROC_STAT_PATH): + """Returns (steal, total) CPU ticks of the whole machine since boot, or None where /proc/stat is unavailable.""" + try: + with open(path, encoding='ascii') as proc_stat: + first_line = proc_stat.readline() + except OSError: + return None + fields = first_line.split() + if not fields or fields[0] != 'cpu' or len(fields) <= STEAL_FIELD_INDEX + 1: + return None + # Only the first 8 counters add up to the total; guest time is already counted in user time + ticks = [int(value) for value in fields[1:STEAL_FIELD_INDEX + 2]] + return ticks[STEAL_FIELD_INDEX], sum(ticks) + + +class CpuUsage: + """Reports CPU use since the previous `sample()` (or since creation, for the first one).""" + + def __init__(self, read_ticks=read_steal_and_total_ticks): + self._read_ticks = read_ticks + self._last_wall = time.monotonic() + self._last_process = time.process_time() + self._last_ticks = read_ticks() + + def sample(self): + """Returns (process percent of one core, steal percent or None) since the previous sample.""" + wall, process, ticks = time.monotonic(), time.process_time(), self._read_ticks() + elapsed = wall - self._last_wall + process_percent = 100 * (process - self._last_process) / elapsed if elapsed > 0 else 0.0 + steal_percent = None + if ticks is not None and self._last_ticks is not None: + total_delta = ticks[1] - self._last_ticks[1] + if total_delta > 0: + steal_percent = 100 * (ticks[0] - self._last_ticks[0]) / total_delta + self._last_wall, self._last_process, self._last_ticks = wall, process, ticks + return process_percent, steal_percent diff --git a/src/racb/core/db.py b/src/racb/core/db.py index c410424..cbd5774 100644 --- a/src/racb/core/db.py +++ b/src/racb/core/db.py @@ -205,3 +205,19 @@ def find_similar_subreddits(subreddit_name, max_results): with conn.cursor() as cur: cur.callproc('find_similar_subreddits', (subreddit_name, max_results)) return [row[0] for row in cur.fetchall()] + + +def save_stream_cursor(cursor_name, next_comment_id, done_blocks): + """Stores where the comment reader stopped: every ID below `next_comment_id`, and the `done_blocks`.""" + conn = get_connection() + with conn.cursor() as cur: + cur.callproc('save_stream_cursor', (cursor_name, next_comment_id, list(done_blocks))) + conn.commit() + + +def get_stream_cursor(cursor_name): + """Returns the saved cursor as a dict (next_comment_id, done_blocks, updated_at), or None if there is none.""" + conn = get_connection() + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.callproc('get_stream_cursor', (cursor_name,)) + return cur.fetchone() diff --git a/src/racb/main.py b/src/racb/main.py index 551bee8..4aed96f 100644 --- a/src/racb/main.py +++ b/src/racb/main.py @@ -15,7 +15,9 @@ from racb.version import __version__ from racb import config, i18n from racb.config import env_bool +from racb.comment_reader import CommentReader from racb.core import db, known_subreddits, reddit, shutdown +from racb.core.cpu_usage import CpuUsage from racb.phases import phase1, phase2, phase3, inbox, cleanup from racb.workers import PeriodicJob, Phase1Workers @@ -38,8 +40,12 @@ class StreamMetrics: """Tracks throughput and events for Reddit comment and inbox streaming. - The stream thread and the Phase 1 workers record events while the heartbeat job reads and resets - the counters, so every access holds a lock. + The stream or reader threads and the Phase 1 workers record events while the heartbeat job reads and + resets the counters, so every access holds a lock. + + Besides the counts, the heartbeat says how the comments were read (requests, comments per request, + seconds per request), how far behind the newest comment the bot is, and how much CPU it used. That tells + whether the bot keeps up with Reddit, and whether the VM's CPU is the limit (see `cpu_usage`). """ def __init__(self): @@ -51,6 +57,17 @@ def __init__(self): self.total_matches = 0 self.total_aux = 0 self.interval_dropped = 0 + self.interval_requests = 0 + self.interval_request_comments = 0 + self.interval_request_seconds = 0.0 + self.interval_head_polls = 0 + self.interval_head_seconds = 0.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 + self.newest_created_utc = None + self.lag_ids = None + self.cpu_usage = CpuUsage() self.last_log_time = time.time() def record_comment(self): @@ -73,9 +90,41 @@ def record_dropped(self): with self._lock: self.interval_dropped += 1 + def record_request(self, seconds, comments_returned): + """Counts one request that read comments (a listing batch or an ID block).""" + with self._lock: + self.interval_requests += 1 + self.interval_request_seconds += seconds + self.interval_request_comments += comments_returned + + def record_head_poll(self, seconds): + with self._lock: + self.interval_head_polls += 1 + self.interval_head_seconds += seconds + + def record_failed_block(self): + """Counts a block of comment IDs given up on after repeated errors.""" + with self._lock: + self.interval_failed_blocks += 1 + + def record_skipped_ids(self, count): + """Counts comment IDs skipped because the reader fell too far behind.""" + with self._lock: + self.interval_skipped_ids += count + + def record_newest_created(self, created_utc): + with self._lock: + if self.newest_created_utc is None or created_utc > self.newest_created_utc: + self.newest_created_utc = created_utc + + def set_lag_ids(self, lag_ids): + with self._lock: + self.lag_ids = lag_ids + def log_heartbeat(self): with self._lock: - elapsed_min = max(1, round((time.time() - self.last_log_time) / 60)) + now = time.time() + elapsed_min = max(1, round((now - self.last_log_time) / 60)) message = ( f'[Heartbeat] Last ~{elapsed_min}m: scanned {self.interval_scanned:,} comments on r/all | ' f'{self.interval_matches} recommendation(s) stored | {self.interval_aux} aux bot reply(s) ' @@ -83,13 +132,55 @@ 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) self.interval_scanned = 0 self.interval_matches = 0 self.interval_aux = 0 self.interval_dropped = 0 - self.last_log_time = time.time() + self.interval_requests = 0 + self.interval_request_comments = 0 + self.interval_request_seconds = 0.0 + self.interval_head_polls = 0 + self.interval_head_seconds = 0.0 + self.interval_failed_blocks = 0 + self.interval_skipped_ids = 0 + self.last_log_time = now logger.info(message) + def _reading_summary(self, now): + """The heartbeat's second half: how comments were read, how far behind, and CPU use.""" + parts = [] + if self.interval_requests: + parts.append( + f'reads: {self.interval_requests:,} requests, ' + f'{self.interval_request_comments / self.interval_requests:.1f} comments/request, ' + f'{self.interval_request_seconds / self.interval_requests:.2f} s/request' + ) + else: + parts.append('reads: 0 requests') + if self.interval_head_polls: + parts.append( + f'head: {self.interval_head_polls:,} polls, ' + f'{self.interval_head_seconds / self.interval_head_polls:.2f} s/poll' + ) + behind = [] + if self.newest_created_utc is not None: + behind.append(f'{max(0, now - self.newest_created_utc):.0f} s') + if self.lag_ids is not None: + behind.append(f'{self.lag_ids:,} IDs') + if behind: + parts.append('behind: ' + ', '.join(behind)) + if self.interval_failed_blocks: + parts.append(f'{self.interval_failed_blocks} ID block(s) failed') + if self.interval_skipped_ids: + parts.append(f'{self.interval_skipped_ids:,} IDs skipped') + process_percent, steal_percent = self.cpu_usage.sample() + cpu = f'CPU: {process_percent:.1f}% of a core' + if steal_percent is not None: + cpu += f', steal {steal_percent:.1f}%' + parts.append(cpu) + return ' | ' + ' | '.join(parts) + metrics = StreamMetrics() @@ -132,12 +223,20 @@ def configure_logging(): logging.getLogger().setLevel(level) -def init_comment_stream(): +def init_comment_stream(continue_after_id=None): + """Returns PRAW's r/all comment stream (STREAM_SOURCE=listing). + + After a stream error, `continue_after_id` (the newest comment seen, as a fullname) makes the new stream + start right after it, instead of skipping everything posted during the back-off. + """ logger.info('Initializing Reddit comment stream...') reddit_instance = reddit.get_reddit_instance() scanned_subreddits = 'all' subreddit = reddit_instance.subreddit(scanned_subreddits) - c_stream = subreddit.stream.comments(skip_existing=True, pause_after=-1) + if continue_after_id: + c_stream = subreddit.stream.comments(continue_after_id=continue_after_id, pause_after=-1) + else: + c_stream = subreddit.stream.comments(skip_existing=True, pause_after=-1) logger.info('Stream initialized successfully. Listening for comments on r/all...') return c_stream @@ -253,8 +352,8 @@ def request_stop(signum, frame): def start_bot(): """Runs the bot until a signal or a fatal error stops it. - The main thread reads the r/all stream. The scheduled jobs and the Phase 1 workers run on their own - threads (`racb.workers`), so they never pause the stream. + The main thread reads new comments (`run_stream`). The scheduled jobs and the Phase 1 workers run on + their own threads (`racb.workers`), so they never pause reading. """ install_signal_handlers() logger.info('Configuring background task schedule...') @@ -265,11 +364,12 @@ def start_bot(): phase1_workers = Phase1Workers(handle_comment_id, worker_count, metrics=metrics).start() logger.info(f'Started {len(jobs)} scheduled job thread(s) and {worker_count} Phase 1 worker thread(s)') + reader_threads = [] try: - run_stream(phase1_workers) + reader_threads = run_stream(phase1_workers) finally: shutdown.request_stop() - wait_for_threads(jobs + phase1_workers.threads, phase1_workers) + wait_for_threads(jobs + phase1_workers.threads + (reader_threads or []), phase1_workers) error = shutdown.fatal_error() if error is not None: @@ -278,12 +378,35 @@ def start_bot(): def run_stream(phase1_workers): + """Reads new comments on the main thread until shutdown, from STREAM_SOURCE. + + Returns the reader threads it started, so the shutdown can wait for them. + """ + if config.stream_source() == 'listing': + run_listing_stream(phase1_workers) + return [] + reader_count = config.stream_reader_count() + reader = CommentReader( + lambda comment: dispatch_comment(comment, phase1_workers), + reader_count, + metrics=metrics, + handle_exception=handle_exception, + ) + logger.info(f'Reading every new comment by ID on {reader_count} reader thread(s)') + reader.start() + reader.run() + return reader.threads + + +def run_listing_stream(phase1_workers): + """STREAM_SOURCE=listing: reads PRAW's r/all comment stream, one request at a time.""" c_stream = None + newest_seen = {'fullname': None} while not shutdown.stop_requested(): try: if c_stream is None: - c_stream = init_comment_stream() - main_loop(c_stream, phase1_workers) + c_stream = init_comment_stream(continue_after_id=newest_seen['fullname']) + main_loop(c_stream, phase1_workers, newest_seen) except Exception as e: should_raise = handle_exception(e) if should_raise: @@ -348,9 +471,10 @@ def handle_exception(e): def dispatch_comment(comment, phase1_workers): - """Handles one stream comment on the stream thread, without any I/O. + """Handles one new comment without any I/O, on the main thread or a reader thread. - Only comments that look like a subreddit suggestion go on to Phase 1, on a worker thread. + Only comments that look like a subreddit suggestion go on to Phase 1, on a worker thread. Reader + threads call this several at once; everything it touches is locked or thread safe. """ try: metrics.record_comment() @@ -395,12 +519,29 @@ def handle_comment_safely(comment): logger.exception(f'Unexpected error while handling comment {reddit.full_url(permalink)}: {e!r}') -def main_loop(c_stream, phase1_workers): - """Reads one batch from the stream. Returns when the batch is done or a shutdown was requested.""" +def main_loop(c_stream, phase1_workers, newest_seen=None): + """Reads one batch from the stream. Returns when the batch is done or a shutdown was requested. + + With `pause_after=-1`, PRAW yields None after every response, so one batch is one request. Its + duration and size go to the heartbeat. `newest_seen['fullname']` keeps the newest comment for a + stream restart. + """ + started = time.monotonic() + batch_size = 0 + newest_created = None for comment in c_stream: if comment is None or shutdown.stop_requested(): break + batch_size += 1 dispatch_comment(comment, phase1_workers) + if newest_seen is not None: + newest_seen['fullname'] = f't1_{comment.id}' + created = vars(comment).get('created_utc') + if created is not None and (newest_created is None or created > newest_created): + newest_created = created + metrics.record_request(time.monotonic() - started, batch_size) + if newest_created is not None: + metrics.record_newest_created(newest_created) if __name__ == '__main__': diff --git a/src/racb/sql/instantiate_db.sql b/src/racb/sql/instantiate_db.sql index fd24677..44838bb 100644 --- a/src/racb/sql/instantiate_db.sql +++ b/src/racb/sql/instantiate_db.sql @@ -189,3 +189,45 @@ BEGIN RETURN QUERY END; $$ LANGUAGE plpgsql; + +-- Where the comment reader (racb/comment_reader.py) stopped, so a restart resumes there instead of +-- skipping what was posted meanwhile. Every comment ID below next_comment_id (a base-36 comment ID as a +-- number) has been read, and so have the blocks of IDs starting at each of done_blocks. +CREATE TABLE IF NOT EXISTS stream_cursors ( + name VARCHAR (32) PRIMARY KEY, + next_comment_id BIGINT NOT NULL, + done_blocks BIGINT[] NOT NULL DEFAULT '{}', + updated_at TIMESTAMP NOT NULL +); + +DROP FUNCTION IF EXISTS save_stream_cursor(VARCHAR, BIGINT, BIGINT[]); +CREATE OR REPLACE FUNCTION save_stream_cursor(cursor_name VARCHAR, new_next_comment_id BIGINT, new_done_blocks BIGINT[]) +RETURNS + VOID +AS +$$ +BEGIN + INSERT INTO stream_cursors(name, next_comment_id, done_blocks, updated_at) + VALUES (cursor_name, new_next_comment_id, new_done_blocks, LOCALTIMESTAMP) + ON CONFLICT (name) DO UPDATE + SET next_comment_id = EXCLUDED.next_comment_id, + done_blocks = EXCLUDED.done_blocks, + updated_at = EXCLUDED.updated_at; +END; +$$ +LANGUAGE plpgsql; + +DROP FUNCTION IF EXISTS get_stream_cursor(VARCHAR); +CREATE OR REPLACE FUNCTION get_stream_cursor(cursor_name VARCHAR) +RETURNS + TABLE (next_comment_id BIGINT, done_blocks BIGINT[], updated_at TIMESTAMP) +AS +$$ +BEGIN + RETURN QUERY + SELECT sc.next_comment_id, sc.done_blocks, sc.updated_at + FROM stream_cursors AS sc + WHERE sc.name = cursor_name; +END; +$$ +LANGUAGE plpgsql; diff --git a/src/racb/workers.py b/src/racb/workers.py index f37e91a..5ea91e9 100644 --- a/src/racb/workers.py +++ b/src/racb/workers.py @@ -1,7 +1,7 @@ """Background threads: the scheduled jobs and the Phase 1 worker pool. -The main thread only reads the r/all comment stream (see `main.py`). Everything slow runs here, because -PRAW's stream only catches up on the newest 100 comments: any pause in reading it loses comments. +The main thread and the comment reader threads only read new comments (see `main.py` and +`comment_reader.py`). Everything slow runs here, so reading never waits for it. Each thread gets its own PRAW instances (`reddit.get_reddit_instance`) and database connection (`db.get_connection`) the first time it uses them. """ @@ -103,16 +103,17 @@ def back_off_after_network_error(error): class Phase1Workers: """A pool of threads that run `handle_comment_id(comment_id)` for matching stream comments. - The stream thread calls `submit()` with comment IDs, not PRAW objects: a comment from the stream is - bound to the stream thread's PRAW instance, so each worker fetches it again through its own. + The stream or reader threads call `submit()` with comment IDs, not PRAW objects: a comment they read + is bound to their own PRAW instance, so each worker fetches it again through its own. """ def __init__(self, handle_comment_id, worker_count, metrics=None, queue_size=PHASE1_QUEUE_SIZE): self.handle_comment_id = handle_comment_id self.metrics = metrics self.queue = queue.Queue(maxsize=queue_size) - # Matches dropped since the queue last had room. Only the stream thread calls submit(), so no lock. + # Matches dropped since the queue last had room. Several reader threads call submit(), hence the lock. self.dropped_since_full = 0 + self._dropped_lock = threading.Lock() self.threads = [ threading.Thread(target=self.work, name=f'phase1-{number}', daemon=True) for number in range(1, worker_count + 1) @@ -129,15 +130,21 @@ def submit(self, comment_id): self.queue.put_nowait(comment_id) except queue.Full: # Logged once when the queue fills and once when it has room again; the heartbeat counts drops - if self.dropped_since_full == 0: - logger.warning(f'Phase 1 queue is full ({self.queue.maxsize}); dropping matches until it has room') - self.dropped_since_full += 1 + with self._dropped_lock: + if self.dropped_since_full == 0: + logger.warning( + f'Phase 1 queue is full ({self.queue.maxsize}); dropping matches until it has room' + ) + self.dropped_since_full += 1 if self.metrics: self.metrics.record_dropped() return False - if self.dropped_since_full: - logger.warning(f'Phase 1 queue has room again; dropped {self.dropped_since_full} match(es) meanwhile') - self.dropped_since_full = 0 + with self._dropped_lock: + if self.dropped_since_full: + logger.warning( + f'Phase 1 queue has room again; dropped {self.dropped_since_full} match(es) meanwhile' + ) + self.dropped_since_full = 0 return True def pending(self): diff --git a/tests/integration/test_reddit_integration.py b/tests/integration/test_reddit_integration.py index 5ef2ad6..02f163a 100644 --- a/tests/integration/test_reddit_integration.py +++ b/tests/integration/test_reddit_integration.py @@ -10,6 +10,7 @@ import prawcore from praw.models import PostMedia +from racb import comment_reader from racb.core import reddit from racb.phases import phase1, phase3, cleanup from scripts.delete_test_artifacts import record_artifact @@ -98,6 +99,27 @@ def test_live_main_startup_flow(monkeypatch): mock_start_bot.assert_called_once() +@pytest.mark.integration +def test_live_comment_reader_reads_recent_comments_by_id(): + """Reads 4 blocks of comment IDs just below r/all's newest comment, one per bot account. + + The comment reader relies on Reddit numbering comments in order across the whole site. If that ever + stops being true, most of these IDs come back empty, this test fails, and CI does not deploy a bot + that would read nothing. + """ + block_count = len(reddit.BOT_NAMES) + first_block = comment_reader.fetch_head() - 10 * comment_reader.BLOCK_SIZE + comments = [] + for index, bot_name in enumerate(reddit.BOT_NAMES): + comments += comment_reader.fetch_block(bot_name, first_block + index * comment_reader.BLOCK_SIZE) + numbers = [comment_reader.comment_id_to_number(comment.id) for comment in comments] + last_id = first_block + block_count * comment_reader.BLOCK_SIZE + assert all(first_block <= number < last_id for number in numbers) + # Comments in private subreddits are not returned, so not every ID comes back + assert len(comments) >= block_count * comment_reader.BLOCK_SIZE // 2, f'only {len(comments)} comments' + assert any(comment_reader.is_shown_on_r_all(comment) for comment in comments) + + @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 new file mode 100644 index 0000000..7a6f364 --- /dev/null +++ b/tests/unit/test_comment_reader.py @@ -0,0 +1,350 @@ +"""Unit tests for racb.comment_reader (reading every new comment by ID block).""" + +import datetime +import threading +import time +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import prawcore +import pytest + +from racb import comment_reader +from racb.comment_reader import ( + BLOCK_SIZE, + CURSOR_NAME, + MAX_BACKLOG_IDS, + CommentReader, + block_fullnames, + comment_id_to_number, + is_shown_on_r_all, + number_to_comment_id, +) +from racb.core import shutdown + +WAIT_SECONDS = 5 + + +def wait_until(condition, timeout=WAIT_SECONDS): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return True + time.sleep(0.01) + return False + + +def fake_comment(number, **fields): + defaults = {'id': number_to_comment_id(number), 'subreddit': 'pics', 'subreddit_type': 'public', + 'over_18': False, 'quarantine': False, 'created_utc': 1_700_000_000.0 + number} + defaults.update(fields) + return SimpleNamespace(**defaults) + + +class FakeCursorStore: + """Stands in for racb.core.db's cursor functions.""" + + def __init__(self, saved=None): + self.saved = saved + self.saves = [] + + def get_stream_cursor(self, name): + assert name == CURSOR_NAME + return self.saved + + def save_stream_cursor(self, name, next_comment_id, done_blocks): + self.saves.append((next_comment_id, list(done_blocks))) + + +def make_reader(head=10_000, saved=None, fetch_block=None, handle_comment=None, metrics=None): + store = FakeCursorStore(saved) + reader = CommentReader( + handle_comment or MagicMock(), + reader_count=2, + metrics=metrics, + fetch_head=lambda: head, + fetch_block=fetch_block or (lambda bot_name, start: []), + cursor_store=store, + ) + return reader, store + + +def queued_blocks(reader): + blocks = [] + while not reader.blocks.empty(): + blocks.append(reader.blocks.get_nowait()) + return blocks + + +@pytest.mark.unit +def test_comment_ids_convert_to_numbers_and_back(): + assert comment_id_to_number('z') == 35 + assert comment_id_to_number('t1_10') == 36 + assert number_to_comment_id(0) == '0' + for comment_id in ('1', 'k3l2x9a', 'zzzzzzz'): + assert number_to_comment_id(comment_id_to_number(comment_id)) == comment_id + + +@pytest.mark.unit +def test_a_block_is_100_consecutive_fullnames(): + fullnames = block_fullnames(comment_id_to_number('a0')) + assert len(fullnames) == BLOCK_SIZE + assert fullnames[0] == 't1_a0' + assert fullnames[1] == 't1_a1' + assert fullnames[-1] == f't1_{number_to_comment_id(comment_id_to_number("a0") + BLOCK_SIZE - 1)}' + + +@pytest.mark.unit +@pytest.mark.parametrize('fields, shown', [ + ({}, True), + ({'subreddit_type': 'restricted'}, True), + ({'over_18': True}, False), + ({'quarantine': True}, False), + ({'subreddit_type': 'private'}, False), + ({'subreddit_type': 'user'}, False), + ({'subreddit': 'u_someone'}, False), + ({'subreddit': SimpleNamespace(display_name='U_Someone')}, False), + ({'subreddit': SimpleNamespace(display_name='pics')}, True), +]) +def test_only_comments_r_all_would_show_are_handled(fields, shown): + assert is_shown_on_r_all(fake_comment(1, **fields)) is shown + + +@pytest.mark.unit +def test_is_shown_on_r_all_never_reads_missing_fields_through_getattr(): + class LazyComment: + def __getattr__(self, name): + raise AssertionError(f'{name} would make PRAW fetch the comment') + + comment = LazyComment() + comment.subreddit = 'pics' + assert is_shown_on_r_all(comment) is True + + +@pytest.mark.unit +def test_a_fresh_start_begins_after_the_newest_comment(): + reader, _ = make_reader(head=10_000) + reader.poll_head() + assert reader.next_id == 10_001 + assert queued_blocks(reader) == [] + + +@pytest.mark.unit +def test_only_complete_blocks_up_to_the_head_are_queued(): + head = {'value': 10_000} + reader, _ = make_reader() + reader.fetch_head = lambda: head['value'] + reader.poll_head() + head['value'] = 10_000 + 2 * BLOCK_SIZE + 50 + reader.poll_head() + assert queued_blocks(reader) == [10_001, 10_001 + BLOCK_SIZE] + assert reader.next_id == 10_001 + 2 * BLOCK_SIZE + + +@pytest.mark.unit +def test_a_restart_resumes_from_the_saved_cursor_and_skips_blocks_already_done(): + start = 10_000 - 3 * BLOCK_SIZE + 1 + saved = {'next_comment_id': start, 'done_blocks': [start + BLOCK_SIZE], + 'updated_at': datetime.datetime(2026, 10, 16, 20, 0)} + reader, _ = make_reader(head=10_000, saved=saved) + reader.poll_head() + assert queued_blocks(reader) == [start, start + 2 * BLOCK_SIZE] + + +@pytest.mark.unit +def test_a_cursor_too_far_behind_is_ignored(): + saved = {'next_comment_id': 1, 'done_blocks': [], 'updated_at': None} + reader, _ = make_reader(head=MAX_BACKLOG_IDS + 10, saved=saved) + reader.poll_head() + assert reader.next_id == MAX_BACKLOG_IDS + 11 + assert queued_blocks(reader) == [] + + +@pytest.mark.unit +def test_a_cursor_that_cannot_be_read_starts_at_the_head(): + reader, store = make_reader(head=500) + store.get_stream_cursor = MagicMock(side_effect=RuntimeError('db down')) + with patch.dict('os.environ', {'DEBUG': 'false'}): + reader.poll_head() + assert reader.next_id == 501 + + +@pytest.mark.unit +def test_the_reader_skips_ahead_when_too_far_behind(): + metrics = MagicMock() + 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() + metrics.record_skipped_ids.assert_called_once_with(MAX_BACKLOG_IDS + 500) + assert reader.next_id == head['value'] + 1 + + +@pytest.mark.unit +def test_the_cursor_holds_the_watermark_and_the_blocks_done_above_it(): + head = {'value': 0} + reader, store = make_reader() + reader.fetch_head = lambda: head['value'] + reader.poll_head() + head['value'] = 3 * BLOCK_SIZE + reader.poll_head() + first, second, third = queued_blocks(reader) + # Blocks finish out of order: the second and third before the first + reader.finish(second) + reader.finish(third) + reader.save_cursor() + assert store.saves[-1] == (first, [second, third]) + reader.finish(first) + reader.save_cursor() + assert store.saves[-1] == (third + BLOCK_SIZE, []) + + +@pytest.mark.unit +def test_read_block_hands_on_shown_comments_and_records_the_request(): + handled = [] + metrics = MagicMock() + comments = [fake_comment(1), fake_comment(2, over_18=True), fake_comment(3)] + fetched = [] + + def fetch_block(bot_name, start): + fetched.append((bot_name, start)) + return comments + + reader, _ = make_reader(fetch_block=fetch_block, handle_comment=handled.append, metrics=metrics) + reader.next_id, reader.in_flight = 200, {100} + reader.read_block('same_post_bot', 100) + assert fetched == [('same_post_bot', 100)] + assert handled == [comments[0], comments[2]] + assert metrics.record_request.call_args.args[1] == 3 + metrics.record_newest_created.assert_called_once_with(comments[2].created_utc) + assert reader.pending() == 0 + + +@pytest.mark.unit +def test_read_block_retries_network_errors(): + attempts = [] + + def flaky_fetch(bot_name, start): + attempts.append(start) + if len(attempts) == 1: + raise prawcore.exceptions.ServerError(MagicMock()) + return [fake_comment(start)] + + handled = [] + reader, _ = make_reader(fetch_block=flaky_fetch, handle_comment=handled.append) + reader.next_id, reader.in_flight = 200, {100} + with patch.dict('os.environ', {'DEBUG': 'false'}), \ + patch('racb.comment_reader.back_off_after_network_error', return_value=False) as mock_back_off: + reader.read_block('AutoCrosspostBot', 100) + assert len(attempts) == 2 + mock_back_off.assert_called_once() + assert len(handled) == 1 + + +@pytest.mark.unit +def test_read_block_gives_up_after_repeated_errors_and_moves_on(): + metrics = MagicMock() + + def failing_fetch(bot_name, start): + raise prawcore.exceptions.ServerError(MagicMock()) + + reader, _ = make_reader(fetch_block=failing_fetch, metrics=metrics) + reader.next_id, reader.in_flight = 200, {100} + with patch.dict('os.environ', {'DEBUG': 'false'}), \ + patch('racb.comment_reader.back_off_after_network_error', return_value=False) as mock_back_off: + reader.read_block('AutoCrosspostBot', 100) + assert mock_back_off.call_count == comment_reader.MAX_BLOCK_ATTEMPTS + metrics.record_failed_block.assert_called_once_with() + assert reader.pending() == 0 + + +@pytest.mark.unit +def test_a_block_interrupted_by_shutdown_stays_unfinished_so_the_restart_reads_it(): + def failing_fetch(bot_name, start): + raise prawcore.exceptions.ServerError(MagicMock()) + + reader, _ = make_reader(fetch_block=failing_fetch) + reader.next_id, reader.in_flight = 200, {100} + with patch.dict('os.environ', {'DEBUG': 'false'}), \ + patch('racb.comment_reader.back_off_after_network_error', return_value=True): + reader.read_block('AutoCrosspostBot', 100) + assert reader.pending() == 1 + + +@pytest.mark.unit +def test_readers_take_turns_between_the_bot_accounts(): + reader = CommentReader(MagicMock(), reader_count=5) + names = [thread._args[0] for thread in reader.threads] + assert names == ['AutoCrosspostBot', 'sub_doesnt_exist_bot', 'same_subreddit_bot', 'same_post_bot', + 'AutoCrosspostBot'] + + +@pytest.mark.unit +def test_reader_threads_read_every_queued_block_then_stop_on_shutdown(): + handled = [] + handled_lock = threading.Lock() + + def handle(comment): + with handled_lock: + handled.append(comment.id) + + def fetch_block(bot_name, start): + return [fake_comment(number) for number in range(start, start + BLOCK_SIZE)] + + head = {'value': 0} + reader, store = make_reader(fetch_block=fetch_block, handle_comment=handle) + reader.fetch_head = lambda: head['value'] + reader.poll_head() + head['value'] = 5 * BLOCK_SIZE + reader.poll_head() + reader.start() + assert wait_until(lambda: reader.pending() == 0) + shutdown.request_stop() + for thread in reader.threads: + thread.join(WAIT_SECONDS) + assert sorted(handled, key=comment_id_to_number) == [number_to_comment_id(n) for n in range(1, 501)] + reader.save_cursor() + assert store.saves[-1] == (501, []) + + +@pytest.mark.unit +def test_run_polls_until_shutdown_and_saves_the_cursor_at_the_end(): + reader, store = make_reader(head=1000) + polls = [] + + def poll_then_stop(): + polls.append(1) + shutdown.request_stop() + return 1000 + + reader.fetch_head = poll_then_stop + reader.run() + assert polls == [1] + assert store.saves[-1] == (1001, []) + + +@pytest.mark.unit +def test_run_hands_head_poll_errors_to_handle_exception(): + errors = [] + + def handle_exception(error): + errors.append(error) + shutdown.request_stop() + return False + + reader, _ = make_reader() + reader.handle_exception = handle_exception + reader.fetch_head = MagicMock(side_effect=prawcore.exceptions.ServerError(MagicMock())) + reader.run() + assert len(errors) == 1 + + +@pytest.mark.unit +def test_a_failing_cursor_save_is_logged_not_raised(): + reader, store = make_reader(head=1000) + reader.poll_head() + store.save_stream_cursor = MagicMock(side_effect=RuntimeError('db down')) + with patch.dict('os.environ', {'DEBUG': 'false'}): + reader.save_cursor() diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 474ab9b..10197cf 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -1,5 +1,6 @@ """Unit tests for racb.config module.""" +import os from unittest.mock import patch import pytest from racb import config @@ -59,3 +60,24 @@ def test_every_bot_has_an_enabled_switch_that_defaults_to_on(): with patch.dict('os.environ', {'SAME_POST_BOT_ENABLED': 'False'}): assert not config.is_bot_enabled(reddit.SAME_POST_BOT_NAME) assert config.is_bot_enabled(reddit.AUTO_CROSSPOST_BOT_NAME) + + +@pytest.mark.unit +@pytest.mark.parametrize('value, expected', [ + (None, 'ids'), ('ids', 'ids'), ('Listing', 'listing'), (' listing ', 'listing'), ('firehose', 'ids'), +]) +def test_stream_source(value, expected): + from racb import config + environment = {} if value is None else {'STREAM_SOURCE': value} + with patch.dict('os.environ', environment): + if value is None: + os.environ.pop('STREAM_SOURCE', None) + assert config.stream_source() == expected + + +@pytest.mark.unit +@pytest.mark.parametrize('value, expected', [('8', 8), ('0', 1), ('many', 4)]) +def test_stream_reader_count(value, expected): + from racb import config + with patch.dict('os.environ', {'STREAM_READERS': value}): + assert config.stream_reader_count() == expected diff --git a/tests/unit/test_cpu_usage.py b/tests/unit/test_cpu_usage.py new file mode 100644 index 0000000..1de848c --- /dev/null +++ b/tests/unit/test_cpu_usage.py @@ -0,0 +1,35 @@ +"""Unit tests for racb.core.cpu_usage.""" + +import pytest + +from racb.core.cpu_usage import CpuUsage, read_steal_and_total_ticks + + +@pytest.mark.unit +def test_read_steal_and_total_ticks_parses_the_cpu_line(tmp_path): + proc_stat = tmp_path / 'stat' + # user nice system idle iowait irq softirq steal guest guest_nice + proc_stat.write_text('cpu 100 0 50 800 10 0 5 35 7 0\ncpu0 1 2 3 4 5 6 7 8 9 10\n') + assert read_steal_and_total_ticks(str(proc_stat)) == (35, 1000) + + +@pytest.mark.unit +def test_read_steal_and_total_ticks_is_none_without_proc_stat(tmp_path): + assert read_steal_and_total_ticks(str(tmp_path / 'missing')) is None + + +@pytest.mark.unit +def test_cpu_usage_reports_steal_since_the_previous_sample(): + readings = iter([(10, 1000), (30, 1200), (30, 1200)]) + usage = CpuUsage(read_ticks=lambda: next(readings)) + process_percent, steal_percent = usage.sample() + assert process_percent >= 0 + assert steal_percent == pytest.approx(10.0) + # No ticks elapsed: no steal figure rather than a division by zero + assert usage.sample()[1] is None + + +@pytest.mark.unit +def test_cpu_usage_without_proc_stat_reports_only_the_process(): + usage = CpuUsage(read_ticks=lambda: None) + assert usage.sample()[1] is None diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 8f46613..8beb83b 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -284,3 +284,91 @@ def stream(): with patch('racb.main.known_subreddits.record'): main_loop(stream(), phase1_workers) phase1_workers.submit.assert_called_once_with('first') + + +@pytest.mark.unit +def test_heartbeat_reports_reads_lag_and_cpu(caplog): + import logging + import time + metrics = StreamMetrics() + metrics.cpu_usage = MagicMock() + metrics.cpu_usage.sample.return_value = (3.25, 1.5) + metrics.record_request(2.0, 100) + metrics.record_request(1.0, 80) + metrics.record_head_poll(4.0) + metrics.record_newest_created(time.time() - 12) + metrics.set_lag_ids(1234) + metrics.record_failed_block() + with caplog.at_level(logging.INFO, logger='racb.main'): + metrics.log_heartbeat() + message = caplog.records[-1].getMessage() + assert 'reads: 2 requests, 90.0 comments/request, 1.50 s/request' in message + assert 'head: 1 polls, 4.00 s/poll' in message + assert 'behind: 12 s, 1,234 IDs' in message + assert '1 ID block(s) failed' in message + assert message.endswith('CPU: 3.2% of a core, steal 1.5%') + assert metrics.interval_requests == 0 + assert metrics.interval_failed_blocks == 0 + + +@pytest.mark.unit +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')] + with patch('racb.main.known_subreddits.record'): + main_loop(iter(comments + [None]), MagicMock(), newest_seen) + assert metrics.interval_requests == requests_before + 1 + assert newest_seen['fullname'] == 't1_second' + + +@pytest.mark.unit +def test_listing_stream_resumes_after_the_newest_comment_after_an_error(): + from racb.main import run_listing_stream + from racb.core import shutdown + calls = [] + + def fake_main_loop(c_stream, phase1_workers, newest_seen): + calls.append(newest_seen['fullname']) + if len(calls) == 1: + newest_seen['fullname'] = 't1_abc' + raise prawcore.exceptions.ServerError(MagicMock()) + shutdown.request_stop() + + with patch('racb.main.init_comment_stream') as mock_init, \ + patch('racb.main.main_loop', side_effect=fake_main_loop), \ + patch('racb.main.handle_exception', return_value=False): + run_listing_stream(MagicMock()) + assert [call.kwargs['continue_after_id'] for call in mock_init.call_args_list] == [None, 't1_abc'] + + +@pytest.mark.unit +def test_init_comment_stream_continues_after_a_given_comment(): + from racb.main import init_comment_stream + with patch('racb.main.reddit.get_reddit_instance') as mock_instance: + init_comment_stream(continue_after_id='t1_abc') + init_comment_stream() + comments = mock_instance.return_value.subreddit.return_value.stream.comments + assert comments.call_args_list[0].kwargs == {'continue_after_id': 't1_abc', 'pause_after': -1} + assert comments.call_args_list[1].kwargs == {'skip_existing': True, 'pause_after': -1} + + +@pytest.mark.unit +@pytest.mark.parametrize('source', ['ids', 'listing']) +def test_run_stream_reads_from_stream_source(source): + from racb.main import run_stream + with patch.dict('os.environ', {'STREAM_SOURCE': source, 'STREAM_READERS': '3'}), \ + patch('racb.main.run_listing_stream') as mock_listing, \ + patch('racb.main.CommentReader') as mock_reader: + threads = run_stream(MagicMock()) + if source == 'listing': + mock_listing.assert_called_once() + mock_reader.assert_not_called() + assert threads == [] + else: + mock_listing.assert_not_called() + assert mock_reader.call_args.args[1] == 3 + mock_reader.return_value.start.assert_called_once_with() + mock_reader.return_value.run.assert_called_once_with() + assert threads is mock_reader.return_value.threads From d33cbeb91e0df7eac8afc73bfd0607e4b3889c51 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 20:16:16 +0000 Subject: [PATCH 2/3] feat: add a live check of the comment reader's assumptions, drop the CPU figures GCP's Always Free tier limits instance hours, disk and egress, not CPU, so the heartbeat's CPU and steal figures are removed. scripts/verify_comment_ids.py checks against live Reddit that comment IDs follow creation time, that ID blocks come back mostly full with the fields the r/all filter reads, how fast IDs are handed out, and what each request costs. Co-Authored-By: Toldry Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FYGhPFNh3ZvVEr2YXt8Qmr --- AGENTS.md | 20 ++- changelog.d/+heartbeat-reading-stats.added.md | 2 +- scripts/verify_comment_ids.py | 146 ++++++++++++++++++ src/racb/core/cpu_usage.py | 56 ------- src/racb/main.py | 13 +- tests/unit/test_cpu_usage.py | 35 ----- tests/unit/test_main.py | 7 +- 7 files changed, 161 insertions(+), 118 deletions(-) create mode 100644 scripts/verify_comment_ids.py delete mode 100644 src/racb/core/cpu_usage.py delete mode 100644 tests/unit/test_cpu_usage.py diff --git a/AGENTS.md b/AGENTS.md index 7fd184c..1ade787 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -210,7 +210,6 @@ A banned target (a 404 whose body says `"reason": "banned"`, `SUBREDDIT_IS_BANNE | :--- | :--- | | [`src/racb/main.py`](./src/racb/main.py) | Application entry point, logging setup, `StreamMetrics` (the heartbeat), `run_stream` (starts the comment reader, or the listing stream loop with `STREAM_SOURCE=listing`), exception recovery, the job list (`build_jobs`), and graceful shutdown on SIGTERM/SIGINT. | | [`src/racb/comment_reader.py`](./src/racb/comment_reader.py) | `CommentReader`: reads every new comment by blocks of 100 consecutive IDs on `STREAM_READERS` threads spread over the 4 accounts, and saves its position (`stream_cursors`) so restarts resume. Its module docstring explains the method. | -| [`src/racb/core/cpu_usage.py`](./src/racb/core/cpu_usage.py) | Process CPU time and VM steal time (`/proc/stat`) between heartbeats. | | [`src/racb/workers.py`](./src/racb/workers.py) | Background threads: `PeriodicJob` (one thread per scheduled job), `Phase1Workers` (the Phase 1 thread pool fed by the stream), and `run_job_safely`. | | [`src/racb/__main__.py`](./src/racb/__main__.py) | Module executable entrypoint for `python -m racb`. | | [`src/racb/__init__.py`](./src/racb/__init__.py) | Package marker; re-exports `__version__`. | @@ -235,6 +234,7 @@ A banned target (a 404 whose body says `"reason": "banned"`, `SUBREDDIT_IS_BANNE | [`scripts/release_level.py`](./scripts/release_level.py) | Used by the CI `release` job: prints `major`, `minor`, `patch` or `none` from the Conventional Commits since the last `v*` tag and whether `changelog.d/` holds fragments. | | [`scripts/cleanup_test_history.py`](./scripts/cleanup_test_history.py) | Deletes integration-test posts and comments from all 4 bot accounts. Supports `--dry-run`, `--limit N` (`0` means all) and `--strict` (exit code 1 if a bot could not be checked or a test item could not be deleted). Slow; for manual use. | | [`scripts/delete_test_artifacts.py`](./scripts/delete_test_artifacts.py) | Deletes only the posts and comments the integration tests recorded in their manifest (`$RACB_TEST_ARTIFACTS_FILE`, default `racb-test-artifacts.jsonl` in the temp directory), as the bot that wrote each one, then removes the manifest. Supports `--dry-run`, `--manifest PATH` and `--strict`. The last step of CI's `integration-tests` job. | +| [`scripts/verify_comment_ids.py`](./scripts/verify_comment_ids.py) | Read-only live check of what `comment_reader.py` assumes: comment IDs follow creation time, 100-ID blocks come back mostly full with the fields the r/all filter reads, how many IDs Reddit hands out per second, and what each request costs (with a rough egress estimate). Needs only `racb.core.reddit`; its docstring shows how to pipe it into the bot container on the VM. | | [`scripts/redeploy.sh`](./scripts/redeploy.sh) | Runs **on the VM**: resets the checkout to a commit of `origin/master` (argument, `SSH_ORIGINAL_COMMAND`, or the latest), runs `docker compose up --detach --build`, and fails if the bot is not running 30 s later. The CI `deploy` job's SSH key is pinned to it. Refuses to run on a dirty or diverged checkout. | | [`scripts/authorize_deploy_key.sh`](./scripts/authorize_deploy_key.sh) | Runs **on the VM** once: adds the CI deploy key to `~/.ssh/authorized_keys` as a forced command for `redeploy.sh` and prints the host key. | | [`scripts/setup_auto_deploy.sh`](./scripts/setup_auto_deploy.sh) | One-time setup of automatic deployment, run in Cloud Shell: service account, Workload Identity Federation, IAP firewall rule, deploy key, GitHub variables. See [`docs/auto-deploy.md`](./docs/auto-deploy.md). | @@ -447,14 +447,13 @@ uv run pytest -m integration - A heartbeat line (`[Heartbeat] Last ~10m: scanned N comments ...`) is logged every 10 minutes by `StreamMetrics`. It is the quickest health and throughput signal. It says `N match(es) dropped, Phase 1 queue full` when the Phase 1 workers fell behind. Besides that, only two lines are logged per backlog: when the queue fills and when it has room again. - Its second half says how comments were read and what it cost, for example - `reads: 412 requests, 87.3 comments/request, 0.95 s/request | head: 160 polls, 1.70 s/poll | behind: 9 s, 340 IDs | CPU: 4.1% of a core, steal 0.6%`: + Its second half says how comments were read, for example + `reads: 412 requests, 87.3 comments/request, 0.95 s/request | head: 50 polls, 1.70 s/poll | behind: 9 s, 340 IDs`: - `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. - `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. - - `CPU`: the process's CPU time as a share of one core, and the VM's steal time (section 8). ### Release & Semantic Versioning Workflow We follow [Semantic Versioning (SemVer)](https://semver.org/) and [Keep a Changelog](https://keepachangelog.com/). **Releases are fully automatic; nobody bumps the version by hand.** @@ -695,13 +694,12 @@ A back-off on the main thread only pauses finding new comments; the readers, wor the image contains `.env`. They `git reset --hard` the VM checkout, so `scripts/redeploy.sh` refuses to run when tracked files were edited on the VM. Keep `redeploy.sh`'s body inside `main()`: the reset rewrites the running script. -- **CPU on the Always Free e2-micro:** the free tier covers the instance itself, not a CPU budget, so CPU use - is not billed; the VM gets about a quarter of a core over time and is throttled beyond that. The heartbeat's - `CPU:` part shows both sides: the process's share of one core (sustained values near 25% mean the bot uses - what the VM gets) and steal (the hypervisor holding the VM back). What the free tier does limit is network - egress, 1 GB a month (https://cloud.google.com/free/docs/free-cloud-features#compute); downloads from Reddit - are ingress, but every request and its TCP acknowledgements go out. Check the VM's network egress in Cloud - Monitoring after a change that adds requests. +- **Always Free limits on the e2-micro** (https://cloud.google.com/free/docs/free-cloud-features#compute): + instance hours, 30 GB-months of standard disk, and 1 GB a month of outbound data transfer. CPU use is not + among them, so it costs nothing; an e2-micro that needs more than its share is only slowed down. Egress is + the limit to watch: downloads from Reddit are inbound, but every request and its TCP acknowledgements go out. + Check the VM's network egress in Cloud Monitoring after a change that adds requests + (`scripts/verify_comment_ids.py` prints a rough estimate for the comment reader). - Python versions: `requires-python = ">=3.11"`. The Docker image, CI and `.python-version` use 3.11, and the unit tests also pass on 3.13. 3.10 is not supported. - Mermaid diagrams: GitHub's renderer rejects unquoted labels that contain parentheses, e.g. diff --git a/changelog.d/+heartbeat-reading-stats.added.md b/changelog.d/+heartbeat-reading-stats.added.md index 22286c0..d27d57b 100644 --- a/changelog.d/+heartbeat-reading-stats.added.md +++ b/changelog.d/+heartbeat-reading-stats.added.md @@ -1 +1 @@ -**Reading and CPU Figures in the Heartbeat:** each heartbeat now also says how many requests read comments, how many comments each returned and how long each took, how far behind the newest comment the bot is, and its CPU use (share of one core, and the VM's steal time), so you can tell whether it keeps up with Reddit and whether the VM is the limit. +**Reading Figures in the Heartbeat:** each heartbeat now also says how many requests read comments, how many comments each returned and how long each took, and how far behind the newest comment the bot is, so you can tell whether it keeps up with Reddit. diff --git a/scripts/verify_comment_ids.py b/scripts/verify_comment_ids.py new file mode 100644 index 0000000..ad5731a --- /dev/null +++ b/scripts/verify_comment_ids.py @@ -0,0 +1,146 @@ +"""Checks, against live Reddit, the assumptions the ID-range comment reader (comment_reader.py) relies on. + +Read-only: it makes about 30 requests as /u/AutoCrosspostBot and writes nothing. It only needs +`racb.core.reddit`, so it also runs on a bot that does not have the comment reader yet. On the VM: + + cd /opt/RedditAutoCrosspostBot + curl -sL https://raw.githubusercontent.com/Toldry/RedditAutoCrosspostBot/claude/project-thread-bf84xq/scripts/verify_comment_ids.py \\ + | docker compose exec -T bot python - + +It checks: +1. Comment IDs follow creation time (the r/all listing's newest 100 comments, sorted by ID, are also + sorted by time, give or take a few seconds). +2. Blocks of 100 consecutive IDs come back mostly full, as one request each, with the fields the r/all + filter reads (over_18, quarantine, subreddit_type). +3. How many comment IDs Reddit hands out per second, and so how many requests per minute the reader needs. +4. How long a block request and an r/all listing request take, and how many bytes each returns. +""" + +import statistics +import sys +import time + +from racb.core import reddit + +BLOCK_SIZE = 100 +SAMPLE_BLOCKS = 10 +# Sample blocks start this many IDs below the newest comment, so every ID in them exists already +SAMPLE_DEPTH_IDS = 5_000 +RATE_SECONDS = 60 +FILTER_FIELDS = ('over_18', 'quarantine', 'subreddit_type', 'created_utc') + +response_bytes = [] + + +def track_response_sizes(instance): + """Records the compressed size of every response (as received over the network) in response_bytes.""" + try: + session = instance._core._requestor._http + except AttributeError: + return + session.hooks['response'].append( + lambda response, *args, **kwargs: response_bytes.append(int(response.headers.get('content-length', 0))) + ) + + +def to_number(comment_id): + return int(comment_id, 36) + + +def to_id(number): + digits = '0123456789abcdefghijklmnopqrstuvwxyz' + text = '' + while number: + number, remainder = divmod(number, 36) + text = digits[remainder] + text + return text or '0' + + +def newest_listing(instance, limit): + started = time.monotonic() + response_bytes.clear() + comments = list(instance.subreddit('all').comments(limit=limit)) + return comments, time.monotonic() - started, sum(response_bytes) + + +def fetch_block(instance, start): + started = time.monotonic() + response_bytes.clear() + fullnames = [f't1_{to_id(number)}' for number in range(start, start + BLOCK_SIZE)] + comments = list(instance.info(fullnames=fullnames)) + return comments, time.monotonic() - started, sum(response_bytes) + + +def verdict(passed, text): + print(f' [{"PASS" if passed else "FAIL"}] {text}') + return passed + + +def main(): + instance = reddit.get_reddit_instance() + track_response_sizes(instance) + results = [] + + print('1. Do comment IDs follow creation time?') + listing, listing_seconds, listing_bytes = newest_listing(instance, 100) + by_id = sorted(listing, key=lambda comment: to_number(comment.id)) + times = [comment.created_utc for comment in by_id] + backwards = max((earlier - later for earlier, later in zip(times, times[1:])), default=0) + print(f' newest r/all comment: {by_id[-1].id}; the newest 100 span {times[-1] - times[0]:.0f} s') + results.append(verdict(backwards <= 10, f'sorted by ID, creation times never go back more than 10 s ' + f'(worst: {backwards:.0f} s)')) + + print(f'\n2. Are blocks of {BLOCK_SIZE} consecutive IDs mostly real comments?') + head = to_number(by_id[-1].id) + counts, seconds, sizes, missing_fields, shown, spans = [], [], [], set(), 0, [] + for index in range(SAMPLE_BLOCKS): + start = head - SAMPLE_DEPTH_IDS - index * BLOCK_SIZE + comments, elapsed, size = fetch_block(instance, start) + counts.append(len(comments)) + seconds.append(elapsed) + sizes.append(size) + for comment in comments: + fields = vars(comment) + missing_fields.update(field for field in FILTER_FIELDS if field not in fields) + hidden = (fields.get('over_18') or fields.get('quarantine') + or fields.get('subreddit_type') in ('private', 'user') + or str(fields.get('subreddit', '')).lower().startswith('u_')) + shown += not hidden + created = [comment.created_utc for comment in comments] + if created: + spans.append(max(created) - min(created)) + total = sum(counts) + print(f' comments per block: {counts}') + results.append(verdict(total >= SAMPLE_BLOCKS * BLOCK_SIZE * 0.5, + f'{total} of {SAMPLE_BLOCKS * BLOCK_SIZE} IDs returned ({total / SAMPLE_BLOCKS:.0f}%)')) + results.append(verdict(not missing_fields, 'every comment has ' + ', '.join(FILTER_FIELDS) + + (f' (missing: {sorted(missing_fields)})' if missing_fields else ''))) + print(f' {shown} of {total} would be on r/all ({100 * shown / max(total, 1):.0f}%)') + if spans: + results.append(verdict(statistics.median(spans) < 120, + f'a block covers {statistics.median(spans):.0f} s of comments (median)')) + + print(f'\n3. How fast does Reddit hand out comment IDs? (waiting {RATE_SECONDS} s)') + time.sleep(RATE_SECONDS) + later, _, _ = newest_listing(instance, 1) + ids_per_second = (to_number(later[0].id) - head) / max(RATE_SECONDS, 1) + blocks_per_minute = ids_per_second * 60 / BLOCK_SIZE + print(f' {ids_per_second:.0f} IDs per second, so about {blocks_per_minute:.0f} block requests per minute') + print(f' (the listing reader saw about 14 comments per second by day)') + + print('\n4. What does each request cost?') + print(f' r/all listing, 100 comments: {listing_seconds:.2f} s, {listing_bytes / 1024:.0f} KiB') + print(f' ID block: median {statistics.median(seconds):.2f} s, max {max(seconds):.2f} s, ' + f'median {statistics.median(sizes) / 1024:.0f} KiB') + if not any(sizes): + print(' (sizes unknown: Reddit sent no content-length)') + # Each block request and its acknowledgements send a few KiB out; an estimate, not a measurement + print(f' rough monthly egress at ~3 KiB out per request: ' + f'{blocks_per_minute * 3 * 60 * 24 * 30 / 1024 / 1024:.1f} GiB (the free tier allows 1 GiB)') + + print(f'\n{sum(results)} of {len(results)} checks passed') + return 0 if all(results) else 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/src/racb/core/cpu_usage.py b/src/racb/core/cpu_usage.py deleted file mode 100644 index f8566a9..0000000 --- a/src/racb/core/cpu_usage.py +++ /dev/null @@ -1,56 +0,0 @@ -"""Measures the bot's CPU use for the heartbeat, so a CPU-starved VM shows up in the log. - -Two numbers per heartbeat interval: -- The process's CPU time (all threads) as a percentage of one core. On the Always Free e2-micro, the - VM is entitled to about a quarter of a core over time, so sustained values near 25% mean the bot is - using up what the VM gets. -- Steal: the share of the VM's CPU time the hypervisor gave to other machines while the VM wanted to run - (`/proc/stat`). High steal means the VM is being throttled, whatever the bot itself uses. In a Docker - container, `/proc/stat` shows the whole VM, which is what matters here. Elsewhere (macOS, Windows) it is - not available, and only the process's own share is reported. -""" - -import time - -PROC_STAT_PATH = '/proc/stat' - -# Field positions on the `cpu` line of /proc/stat: user nice system idle iowait irq softirq steal ... -STEAL_FIELD_INDEX = 7 - - -def read_steal_and_total_ticks(path=PROC_STAT_PATH): - """Returns (steal, total) CPU ticks of the whole machine since boot, or None where /proc/stat is unavailable.""" - try: - with open(path, encoding='ascii') as proc_stat: - first_line = proc_stat.readline() - except OSError: - return None - fields = first_line.split() - if not fields or fields[0] != 'cpu' or len(fields) <= STEAL_FIELD_INDEX + 1: - return None - # Only the first 8 counters add up to the total; guest time is already counted in user time - ticks = [int(value) for value in fields[1:STEAL_FIELD_INDEX + 2]] - return ticks[STEAL_FIELD_INDEX], sum(ticks) - - -class CpuUsage: - """Reports CPU use since the previous `sample()` (or since creation, for the first one).""" - - def __init__(self, read_ticks=read_steal_and_total_ticks): - self._read_ticks = read_ticks - self._last_wall = time.monotonic() - self._last_process = time.process_time() - self._last_ticks = read_ticks() - - def sample(self): - """Returns (process percent of one core, steal percent or None) since the previous sample.""" - wall, process, ticks = time.monotonic(), time.process_time(), self._read_ticks() - elapsed = wall - self._last_wall - process_percent = 100 * (process - self._last_process) / elapsed if elapsed > 0 else 0.0 - steal_percent = None - if ticks is not None and self._last_ticks is not None: - total_delta = ticks[1] - self._last_ticks[1] - if total_delta > 0: - steal_percent = 100 * (ticks[0] - self._last_ticks[0]) / total_delta - self._last_wall, self._last_process, self._last_ticks = wall, process, ticks - return process_percent, steal_percent diff --git a/src/racb/main.py b/src/racb/main.py index 4aed96f..0813e4d 100644 --- a/src/racb/main.py +++ b/src/racb/main.py @@ -17,7 +17,6 @@ from racb.config import env_bool from racb.comment_reader import CommentReader from racb.core import db, known_subreddits, reddit, shutdown -from racb.core.cpu_usage import CpuUsage from racb.phases import phase1, phase2, phase3, inbox, cleanup from racb.workers import PeriodicJob, Phase1Workers @@ -44,8 +43,8 @@ class StreamMetrics: resets the counters, so every access holds a lock. Besides the counts, the heartbeat says how the comments were read (requests, comments per request, - seconds per request), how far behind the newest comment the bot is, and how much CPU it used. That tells - whether the bot keeps up with Reddit, and whether the VM's CPU is the limit (see `cpu_usage`). + seconds per request) and how far behind the newest comment the bot is, which tells whether it keeps up + with Reddit. """ def __init__(self): @@ -67,7 +66,6 @@ def __init__(self): # Creation time of the newest comment read so far, and how many IDs wait to be read self.newest_created_utc = None self.lag_ids = None - self.cpu_usage = CpuUsage() self.last_log_time = time.time() def record_comment(self): @@ -148,7 +146,7 @@ def log_heartbeat(self): logger.info(message) def _reading_summary(self, now): - """The heartbeat's second half: how comments were read, how far behind, and CPU use.""" + """The heartbeat's second half: how comments were read, and how far behind.""" parts = [] if self.interval_requests: parts.append( @@ -174,11 +172,6 @@ def _reading_summary(self, now): parts.append(f'{self.interval_failed_blocks} ID block(s) failed') if self.interval_skipped_ids: parts.append(f'{self.interval_skipped_ids:,} IDs skipped') - process_percent, steal_percent = self.cpu_usage.sample() - cpu = f'CPU: {process_percent:.1f}% of a core' - if steal_percent is not None: - cpu += f', steal {steal_percent:.1f}%' - parts.append(cpu) return ' | ' + ' | '.join(parts) diff --git a/tests/unit/test_cpu_usage.py b/tests/unit/test_cpu_usage.py deleted file mode 100644 index 1de848c..0000000 --- a/tests/unit/test_cpu_usage.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Unit tests for racb.core.cpu_usage.""" - -import pytest - -from racb.core.cpu_usage import CpuUsage, read_steal_and_total_ticks - - -@pytest.mark.unit -def test_read_steal_and_total_ticks_parses_the_cpu_line(tmp_path): - proc_stat = tmp_path / 'stat' - # user nice system idle iowait irq softirq steal guest guest_nice - proc_stat.write_text('cpu 100 0 50 800 10 0 5 35 7 0\ncpu0 1 2 3 4 5 6 7 8 9 10\n') - assert read_steal_and_total_ticks(str(proc_stat)) == (35, 1000) - - -@pytest.mark.unit -def test_read_steal_and_total_ticks_is_none_without_proc_stat(tmp_path): - assert read_steal_and_total_ticks(str(tmp_path / 'missing')) is None - - -@pytest.mark.unit -def test_cpu_usage_reports_steal_since_the_previous_sample(): - readings = iter([(10, 1000), (30, 1200), (30, 1200)]) - usage = CpuUsage(read_ticks=lambda: next(readings)) - process_percent, steal_percent = usage.sample() - assert process_percent >= 0 - assert steal_percent == pytest.approx(10.0) - # No ticks elapsed: no steal figure rather than a division by zero - assert usage.sample()[1] is None - - -@pytest.mark.unit -def test_cpu_usage_without_proc_stat_reports_only_the_process(): - usage = CpuUsage(read_ticks=lambda: None) - assert usage.sample()[1] is None diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 8beb83b..ad68114 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -287,12 +287,10 @@ def stream(): @pytest.mark.unit -def test_heartbeat_reports_reads_lag_and_cpu(caplog): +def test_heartbeat_reports_reads_and_lag(caplog): import logging import time metrics = StreamMetrics() - metrics.cpu_usage = MagicMock() - metrics.cpu_usage.sample.return_value = (3.25, 1.5) metrics.record_request(2.0, 100) metrics.record_request(1.0, 80) metrics.record_head_poll(4.0) @@ -305,8 +303,7 @@ def test_heartbeat_reports_reads_lag_and_cpu(caplog): assert 'reads: 2 requests, 90.0 comments/request, 1.50 s/request' in message assert 'head: 1 polls, 4.00 s/poll' in message assert 'behind: 12 s, 1,234 IDs' in message - assert '1 ID block(s) failed' in message - assert message.endswith('CPU: 3.2% of a core, steal 1.5%') + assert message.endswith('1 ID block(s) failed') assert metrics.interval_requests == 0 assert metrics.interval_failed_blocks == 0 From d21e8f814f402e782d7815684cfb45afa842a008 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 20:25:20 +0000 Subject: [PATCH 3/3] fix: drop NSFW comments by their post and make the head poll robust The live check showed /api/info comments carry no over_18 or quarantine field, so the reader's NSFW filter did nothing. Phase 1 now drops matches on NSFW or quarantined posts, reading the submission it already fetches. The r/all listing sometimes returns nothing for limit=1, which would have stopped the head poll; it now asks for 25 and takes the newest. Co-Authored-By: Toldry Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FYGhPFNh3ZvVEr2YXt8Qmr --- AGENTS.md | 12 +++++++---- scripts/verify_comment_ids.py | 35 +++++++++++++++++++++++-------- src/racb/comment_reader.py | 24 ++++++++++++++------- src/racb/phases/phase1.py | 14 +++++++++++++ tests/unit/test_comment_reader.py | 29 +++++++++++++++++++++---- tests/unit/test_phase1.py | 22 +++++++++++++++++++ 6 files changed, 111 insertions(+), 25 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 615c100..778bf75 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -90,7 +90,9 @@ flowchart TD R -->|No| X1["ignore"] R -->|Yes| M{"distinguished ==\n'moderator'?"} M -->|Yes| X2["ignore"] - M -->|No| T{"Submission title contains\nsub/subs/subreddit/subreddits?"} + M -->|No| NS{"Submission NSFW\nor quarantined?"} + NS -->|Yes| X6["ignore"] + NS -->|No| T{"Submission title contains\nsub/subs/subreddit/subreddits?"} T -->|Yes| X3["ignore"] T -->|No| BL{"Source or target in\nSUB_BLACKLIST?\n(case-insensitive)"} BL -->|Yes| X4["ignore"] @@ -116,8 +118,10 @@ A banned target (a 404 whose body says `"reason": "banned"`, `SUBREDDIT_IS_BANNE - **Phase 1 ([`src/racb/phases/phase1.py`](./src/racb/phases/phase1.py)):** - Continuous reading of every new comment by ID (`comment_reader.CommentReader`, the default, `STREAM_SOURCE=ids`): blocks of 100 consecutive comment IDs fetched with `reddit.info()` on - `STREAM_READERS` threads, keeping only what r/all would show (`is_shown_on_r_all`: not NSFW, quarantined, - private or on a user profile). Subreddits that opted out of r/all cannot be told apart, so their comments + `STREAM_READERS` threads, keeping only what r/all could show (`is_shown_on_r_all`: not private or on a user + profile). `/api/info` comments have no `over_18` or `quarantine` field, so Phase 1 drops NSFW and + quarantined ones by their submission (`phase1.is_hidden_from_r_all`, no extra request). The head poll + asks the listing for 25 comments and takes the newest, because `limit=1` sometimes returns none. Subreddits that opted out of r/all cannot be told apart, so their comments are read too. `STREAM_SOURCE=listing` instead reads `reddit.subreddit('all').stream.comments(skip_existing=True, pause_after=-1)` one request at a time, which sees only a fraction of r/all (section 9). @@ -724,7 +728,7 @@ A back-off on the main thread only pauses finding new comments; the readers, wor | Phase 1 backlog | Phase 1 (duplicate search, up to 200 preview downloads per match, subreddit suggestions) runs on `PHASE1_WORKERS` threads. If matches arrive faster than they can handle for long enough to fill the 1000-ID queue, new matches are dropped; the heartbeat counts them. Not observed; matches are a small fraction of `r/all`. | | Comment coverage | `STREAM_SOURCE=listing` reads at most 100 comments per request, one request at a time. On 2026-10-15/16 that was about 850 comments a minute by day and up to 2,500 at night, when Reddit answered faster, so it missed most of r/all. The `ids` reader was written for that; it relies on Reddit numbering comments in order site-wide, which `test_live_comment_reader_reads_recent_comments_by_id` checks before every deploy, and its real coverage is still to be confirmed from the heartbeat. | | Re-reads after a crash | The comment cursor is saved every 30 s and at shutdown. After a crash or `kill -9`, up to 30 s of comments are read again, so an auxiliary bot could reply twice to a comment from those seconds. | -| Opted-out subreddits | `/api/info` does not say whether a subreddit opted out of r/all, so the `ids` reader also reads comments in subreddits that did (NSFW, quarantined, private and profile comments are dropped). | +| Opted-out subreddits | `/api/info` does not say whether a subreddit opted out of r/all, so the `ids` reader also reads comments in subreddits that did (private and profile comments are dropped by the reader, NSFW and quarantined ones by Phase 1). | | Rotated reposts | Not detected as duplicates (see the image-matching note in section 5). | | Edited video reposts | Videos are compared by their preview frame only. A trimmed, re-cut or re-encoded copy usually has a different preview frame and is missed. | | Pre-v2.3 rows | Have `target_subreddit = NULL`, so the bait-and-switch check is skipped for them until they age out (at most one Phase 3 waiting period after upgrading). | diff --git a/scripts/verify_comment_ids.py b/scripts/verify_comment_ids.py index ad5731a..3770c6d 100644 --- a/scripts/verify_comment_ids.py +++ b/scripts/verify_comment_ids.py @@ -1,6 +1,6 @@ """Checks, against live Reddit, the assumptions the ID-range comment reader (comment_reader.py) relies on. -Read-only: it makes about 30 requests as /u/AutoCrosspostBot and writes nothing. It only needs +Read-only: it makes about 15 requests as /u/AutoCrosspostBot and writes nothing. It only needs `racb.core.reddit`, so it also runs on a bot that does not have the comment reader yet. On the VM: cd /opt/RedditAutoCrosspostBot @@ -11,7 +11,8 @@ 1. Comment IDs follow creation time (the r/all listing's newest 100 comments, sorted by ID, are also sorted by time, give or take a few seconds). 2. Blocks of 100 consecutive IDs come back mostly full, as one request each, with the fields the r/all - filter reads (over_18, quarantine, subreddit_type). + filter reads (subreddit_type), and their submissions say whether they are NSFW or quarantined (over_18, + quarantine), which Phase 1 reads since the comments themselves do not. 3. How many comment IDs Reddit hands out per second, and so how many requests per minute the reader needs. 4. How long a block request and an r/all listing request take, and how many bytes each returns. """ @@ -27,7 +28,10 @@ # Sample blocks start this many IDs below the newest comment, so every ID in them exists already SAMPLE_DEPTH_IDS = 5_000 RATE_SECONDS = 60 -FILTER_FIELDS = ('over_18', 'quarantine', 'subreddit_type', 'created_utc') +FILTER_FIELDS = ('subreddit_type', 'created_utc') +SUBMISSION_FIELDS = ('over_18', 'quarantine') +# The r/all listing is filtered after the limit is applied, so a limit of 1 can return nothing +HEAD_LISTING_LIMIT = 25 response_bytes = [] @@ -92,7 +96,7 @@ def main(): print(f'\n2. Are blocks of {BLOCK_SIZE} consecutive IDs mostly real comments?') head = to_number(by_id[-1].id) - counts, seconds, sizes, missing_fields, shown, spans = [], [], [], set(), 0, [] + counts, seconds, sizes, missing_fields, shown, spans, link_ids = [], [], [], set(), 0, [], [] for index in range(SAMPLE_BLOCKS): start = head - SAMPLE_DEPTH_IDS - index * BLOCK_SIZE comments, elapsed, size = fetch_block(instance, start) @@ -102,10 +106,11 @@ def main(): for comment in comments: fields = vars(comment) missing_fields.update(field for field in FILTER_FIELDS if field not in fields) - hidden = (fields.get('over_18') or fields.get('quarantine') - or fields.get('subreddit_type') in ('private', 'user') + hidden = (fields.get('subreddit_type') in ('private', 'user') or str(fields.get('subreddit', '')).lower().startswith('u_')) shown += not hidden + if not hidden and fields.get('link_id'): + link_ids.append(fields['link_id']) created = [comment.created_utc for comment in comments] if created: spans.append(max(created) - min(created)) @@ -115,15 +120,27 @@ def main(): f'{total} of {SAMPLE_BLOCKS * BLOCK_SIZE} IDs returned ({total / SAMPLE_BLOCKS:.0f}%)')) results.append(verdict(not missing_fields, 'every comment has ' + ', '.join(FILTER_FIELDS) + (f' (missing: {sorted(missing_fields)})' if missing_fields else ''))) - print(f' {shown} of {total} would be on r/all ({100 * shown / max(total, 1):.0f}%)') + print(f' {shown} of {total} pass the comment filter ({100 * shown / max(total, 1):.0f}%)') + # One request for up to 100 of their submissions, to see what Phase 1's NSFW check reads + submissions = list(instance.info(fullnames=list(dict.fromkeys(link_ids))[:BLOCK_SIZE])) + missing_submission_fields = {field for submission in submissions for field in SUBMISSION_FIELDS + if field not in vars(submission)} + results.append(verdict(submissions and not missing_submission_fields, + f'{len(submissions)} submissions fetched, each has ' + ', '.join(SUBMISSION_FIELDS) + + (f' (missing: {sorted(missing_submission_fields)})' if missing_submission_fields else ''))) + hidden_posts = sum(bool(vars(s).get('over_18') or vars(s).get('quarantine')) for s in submissions) + print(f' {hidden_posts} of {len(submissions)} of those posts are NSFW or quarantined') if spans: results.append(verdict(statistics.median(spans) < 120, f'a block covers {statistics.median(spans):.0f} s of comments (median)')) print(f'\n3. How fast does Reddit hand out comment IDs? (waiting {RATE_SECONDS} s)') time.sleep(RATE_SECONDS) - later, _, _ = newest_listing(instance, 1) - ids_per_second = (to_number(later[0].id) - head) / max(RATE_SECONDS, 1) + later, _, _ = newest_listing(instance, HEAD_LISTING_LIMIT) + if not later: + print(' the r/all listing returned no comments; cannot measure') + later = by_id + ids_per_second = (max(to_number(comment.id) for comment in later) - head) / max(RATE_SECONDS, 1) blocks_per_minute = ids_per_second * 60 / BLOCK_SIZE print(f' {ids_per_second:.0f} IDs per second, so about {blocks_per_minute:.0f} block requests per minute') print(f' (the listing reader saw about 14 comments per second by day)') diff --git a/src/racb/comment_reader.py b/src/racb/comment_reader.py index 0149b59..8595acf 100644 --- a/src/racb/comment_reader.py +++ b/src/racb/comment_reader.py @@ -17,9 +17,11 @@ saved to the database every `CURSOR_SAVE_SECONDS` and at shutdown, so a restart resumes where the previous run stopped instead of skipping what was posted meanwhile. -`/api/info` also returns comments that r/all leaves out (private and NSFW subreddits, user profiles), so -those are dropped. Subreddits that only opted out of r/all cannot be told apart from a comment, so their -comments are now read too. +`/api/info` also returns comments that r/all leaves out (private subreddits, user profiles), so those are +dropped. Its comments say nothing about NSFW or quarantine (no `over_18` or `quarantine` field, checked live +on 2026-10-16), so Phase 1 drops those by their submission (`phase1.is_hidden_from_r_all`), which it fetches +anyway, and only for regex matches. Subreddits that only opted out of r/all cannot be told apart from a +comment, so their comments are now read too. """ import logging @@ -36,6 +38,10 @@ # /api/info returns at most 100 items per request BLOCK_SIZE = 100 +# Comments the head poll asks for. Reddit filters the r/all listing after applying the limit, so asking for +# one comment sometimes returns none (seen live on 2026-10-16); the newest of a few is the head +HEAD_LISTING_LIMIT = 25 + # Seconds to wait between two head polls (each poll itself takes a few seconds). It only sets how soon a new # comment is read, not how many: every block up to the head is queued at once. Fewer polls, fewer requests. HEAD_POLL_SECONDS = 10 @@ -81,11 +87,13 @@ def block_fullnames(block_start): def is_shown_on_r_all(comment): - """Whether r/all would list this comment: not NSFW, quarantined, private or on a user profile.""" + """Whether r/all could list this comment: not in a private subreddit or on a user profile. + + NSFW and quarantined comments pass: `/api/info` comments carry no field for either, so Phase 1 checks + their submission instead (`phase1.is_hidden_from_r_all`). + """ # vars(), not getattr: reading a field the comment lacks would make PRAW fetch it, one request each fields = vars(comment) - if fields.get('over_18') or fields.get('quarantine'): - return False if fields.get('subreddit_type') in HIDDEN_SUBREDDIT_TYPES: return False subreddit = fields.get('subreddit') @@ -95,10 +103,10 @@ def is_shown_on_r_all(comment): def fetch_head(): """Returns the number of the newest comment on r/all (one request, as /u/AutoCrosspostBot).""" - newest = list(reddit.get_reddit_instance().subreddit('all').comments(limit=1)) + newest = list(reddit.get_reddit_instance().subreddit('all').comments(limit=HEAD_LISTING_LIMIT)) if not newest: raise RuntimeError('The r/all comment listing returned no comments') - return comment_id_to_number(newest[0].id) + return max(comment_id_to_number(comment.id) for comment in newest) def fetch_block(bot_name, block_start): diff --git a/src/racb/phases/phase1.py b/src/racb/phases/phase1.py index 5e0917a..2ebc114 100644 --- a/src/racb/phases/phase1.py +++ b/src/racb/phases/phase1.py @@ -24,6 +24,10 @@ def handle_incoming_comment(comment, metrics=None): if is_mod_post(comment): return + if is_hidden_from_r_all(comment): + logger.debug(f'Ignoring a comment on an NSFW or quarantined post: {reddit.full_url(comment.permalink)}') + return + if title_contains_prohibited_phrases(comment): return @@ -84,6 +88,16 @@ def is_mod_post(comment): return comment.distinguished == 'moderator' +def is_hidden_from_r_all(comment): + """Whether the comment's post is NSFW or quarantined, which r/all never shows. + + The comment reader cannot tell (`/api/info` comments have neither field), so this reads the submission, + which the title check fetches anyway. + """ + submission = comment.submission + return bool(getattr(submission, 'over_18', False) or getattr(submission, 'quarantine', False)) + + def is_top_level_comment(comment): return comment.parent_id == comment.link_id diff --git a/tests/unit/test_comment_reader.py b/tests/unit/test_comment_reader.py index 7a6f364..8773c97 100644 --- a/tests/unit/test_comment_reader.py +++ b/tests/unit/test_comment_reader.py @@ -35,8 +35,9 @@ def wait_until(condition, timeout=WAIT_SECONDS): def fake_comment(number, **fields): + # The fields /api/info returns for a comment (it has no over_18 or quarantine) defaults = {'id': number_to_comment_id(number), 'subreddit': 'pics', 'subreddit_type': 'public', - 'over_18': False, 'quarantine': False, 'created_utc': 1_700_000_000.0 + number} + 'created_utc': 1_700_000_000.0 + number} defaults.update(fields) return SimpleNamespace(**defaults) @@ -98,8 +99,9 @@ def test_a_block_is_100_consecutive_fullnames(): @pytest.mark.parametrize('fields, shown', [ ({}, True), ({'subreddit_type': 'restricted'}, True), - ({'over_18': True}, False), - ({'quarantine': True}, False), + # Phase 1 drops these by their submission: /api/info comments carry neither field + ({'over_18': True}, True), + ({'quarantine': True}, True), ({'subreddit_type': 'private'}, False), ({'subreddit_type': 'user'}, False), ({'subreddit': 'u_someone'}, False), @@ -121,6 +123,25 @@ def __getattr__(self, name): assert is_shown_on_r_all(comment) is True +@pytest.mark.unit +def test_fetch_head_takes_the_newest_of_several_comments(): + listing = [SimpleNamespace(id='k3l2x9a'), SimpleNamespace(id='k3l2x9c'), SimpleNamespace(id='k3l2x9b')] + 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() == comment_id_to_number('k3l2x9c') + assert instance.subreddit.return_value.comments.call_args.kwargs['limit'] > 1 + + +@pytest.mark.unit +def test_fetch_head_raises_on_an_empty_listing(): + instance = MagicMock() + instance.subreddit.return_value.comments.return_value = iter([]) + with patch('racb.comment_reader.reddit.get_reddit_instance', return_value=instance): + with pytest.raises(RuntimeError): + comment_reader.fetch_head() + + @pytest.mark.unit def test_a_fresh_start_begins_after_the_newest_comment(): reader, _ = make_reader(head=10_000) @@ -205,7 +226,7 @@ def test_the_cursor_holds_the_watermark_and_the_blocks_done_above_it(): def test_read_block_hands_on_shown_comments_and_records_the_request(): handled = [] metrics = MagicMock() - comments = [fake_comment(1), fake_comment(2, over_18=True), fake_comment(3)] + comments = [fake_comment(1), fake_comment(2, subreddit_type='private'), fake_comment(3)] fetched = [] def fetch_block(bot_name, start): diff --git a/tests/unit/test_phase1.py b/tests/unit/test_phase1.py index 9932d1b..e0a6ada 100644 --- a/tests/unit/test_phase1.py +++ b/tests/unit/test_phase1.py @@ -55,6 +55,28 @@ def test_title_contains_prohibited_phrases(make_mock_comment, make_mock_submissi assert phase1.title_contains_prohibited_phrases(make_mock_comment(submission=good_title_sub)) is False +@pytest.mark.unit +@pytest.mark.parametrize('fields, hidden', [ + ({}, False), + ({'over_18': True}, True), + ({'quarantine': True}, True), +]) +def test_is_hidden_from_r_all_reads_the_submission(make_mock_comment, make_mock_submission, fields, hidden): + submission = make_mock_submission(over_18=fields.get('over_18', False)) + if 'quarantine' in fields: + submission.quarantine = fields['quarantine'] + assert phase1.is_hidden_from_r_all(make_mock_comment(submission=submission)) is hidden + + +@pytest.mark.unit +def test_handle_incoming_comment_ignores_nsfw_posts(make_mock_comment, make_mock_submission): + comment = make_mock_comment(body='r/test_source', subreddit='test_source', + submission=make_mock_submission(over_18=True)) + with patch('racb.phases.phase1.reply_to_source_equals_target_comment') as mock_reply: + assert phase1.handle_incoming_comment(comment) is None + mock_reply.assert_not_called() + + @pytest.mark.unit def test_is_subreddit_name_length_valid(): assert phase1.is_subreddit_name_length_valid('abc') is True