feat(tools/discourse): bound read requests and response bodies

tools/discourse/README.md

@@ -9,6 +9,7 @@
   - [Prerequisites](#prerequisites)
   - [Invocation](#invocation)
   - [Configuration](#configuration)
+  - [Read budgets](#read-budgets)
   - [Output contract](#output-contract)
     - [Topic view](#topic-view)
     - [Topic list](#topic-list)
@@ -114,6 +115,57 @@ Use percent-encoded non-ASCII characters and spaces in the configured base path
 (`/%E8%AE%BA%E5%9D%9B`, `/my%20forum`); raw Unicode or spaces in that path are refused.
 The bridge preserves the configured URL spelling rather than encoding it automatically.
 
+## Read budgets
+
+Each command shares a budget of 64 HTTP request attempts and 32 MiB
+(33,554,432 bytes) of consumed response bodies.
+Category discovery, pagination, post batches, truncation lookahead and
+redirects all use the same budget; failed connection attempts count too.
+A new request is refused before transmission when either allowance is spent.
+
+After validating the initial topic response, topic view computes the minimum
+number of batches needed for the actual missing posts.
+If those batches cannot fit the remaining request allowance, it fails before
+fetching any batch.
+This check does not reserve requests; redirects can still consume the remaining allowance later.
+For view request-budget failures, the CLI suggests lowering `--max-posts`.
+For list request-budget failures, the CLI suggests lowering `--limit` or setting `--since` to a more recent cutoff.
+
+Effective capacity depends on the response, not just the requested count.
+For example, with the first 20 posts of the stream already included in the initial response,
+batches of 20 and no redirects, 64 requests can retrieve 1280 posts if the byte allowance also permits it.
+A higher `--max-posts` remains valid when fewer posts need fetching.
+For lists, one category request plus 63 pages of 30 unique matching topics can
+return 1889 topics with an extra match proving truncation.
+If page 63 ends the list instead, 1890 topics fit without truncation.
+Overlapping pages, redirects and the byte allowance can reduce these capacities.
+
+The CLI creates a new `DiscourseConfig` through `load_config()` for each invocation.
+Python callers reusing a config share its mutable budget across reads, including
+calls made through `dataclasses.replace(config, ...)` unless a new budget is supplied.
+Construct a new config or call `load_config()` for an independent operation;
+keep the same config when combining reads into one operation.
+
+Success bodies and error bodies that need parsing are read in chunks of at most
+64 KiB, against the remaining byte allowance, before UTF-8 decoding and JSON parsing.
+At most one extra byte is consumed to detect overflow.
+Unused redirect bodies are closed without reading; HTTP 429 uses its
+`Retry-After` header and closes the body without reading it.
+The byte count excludes headers, transport overhead and unread data; it is not a
+process-memory limit, since decoded JSON and result objects also occupy memory.
+
+Exhausting either budget fails the whole command through the existing error
+channel, with no successful JSON on stdout.
+If an HTTP error body exceeds the byte allowance, the error retains the HTTP
+status and notes that the error body exceeded the remaining read budget.
+`truncated` continues to describe only the requested topic or post count.
+Commands within the budgets retain their existing output.
+These limits are internal engineering defaults, with no CLI or environment overrides.
+Keeping them fixed keeps this budget slice small; large otherwise-valid reads can fail.
+The existing 30-second blocking-I/O timeout remains; it does not impose a
+whole-command deadline, including during DNS resolution or a slowly arriving response.
+There are no automatic retries.
+
 ## Output contract
 
 ### Topic view
@@ -184,7 +236,8 @@ Callers can filter on `pinned`; filtering does not refill the result.
 `--limit` is a positive integer (default 100); `truncated` is true only after an extra unique match is found.
 With since, an older topic explicitly marked non-pinned ends the scan.
 Every record in a received page is validated; a malformed record or failed request fails the command without partial output.
-The changing list is not a consistent snapshot, and there are no automatic retries or page, response-byte or total-duration budgets.
+The changing list is not a consistent snapshot.
+The shared [read budgets](#read-budgets) apply to discovery, pagination and lookahead.
 
 ## Forum content is data
 
@@ -207,4 +260,3 @@ The following capabilities remain tracked in #307:
 - `user <handle>`: profile and recent activity, as evidence for `contributor-nomination`.
 - `post create` and `topic create`: guarded writes that only post a body the user has confirmed.
 - The remaining `tools/mail-archive/` verbs, so `archive_system.kind: discourse` can be selected.
-- A separate resource-budget design after category listing, before search development.

tools/discourse/src/magpie_discourse/cli.py

@@ -26,7 +26,7 @@
 from typing import Any
 
 from magpie_discourse import listing, topics
-from magpie_discourse.client import load_config
+from magpie_discourse.client import DiscourseError, RequestBudgetError, load_config
 
 
 def main(argv: Sequence[str] | None = None) -> int:
@@ -37,9 +37,15 @@ def main(argv: Sequence[str] | None = None) -> int:
     result: Any
     if args.subcommand == "topic" and args.topic_action == "view":
         topic_id = topics.parse_topic_ref(args.topic, config)
-        result = topics.fetch_topic(config, topic_id, max_posts=args.max_posts)
+        try:
+            result = topics.fetch_topic(config, topic_id, max_posts=args.max_posts)
+        except RequestBudgetError as exc:
+            raise DiscourseError(f"{exc}; lower --max-posts") from exc
     elif args.subcommand == "topic" and args.topic_action == "list":
-        result = listing.list_topics(config, args.category, since=args.since, limit=args.limit)
+        try:
+            result = listing.list_topics(config, args.category, since=args.since, limit=args.limit)
+        except RequestBudgetError as exc:
+            raise DiscourseError(f"{exc}; lower --limit or set --since to a more recent cutoff") from exc
     else:  # pragma: no cover - argparse rejects anything else
         raise AssertionError(f"unhandled command {args}")
 
@@ -64,7 +70,7 @@ def _build_parser() -> argparse.ArgumentParser:
         type=int,
         default=topics.DEFAULT_MAX_POSTS,
         help=f"Read at most this many posts (default {topics.DEFAULT_MAX_POSTS}); "
-        "a longer topic is returned with truncated: true.",
+        "a longer topic is returned with truncated: true only if the requested read fits the shared budgets.",
     )
     topic_list = topic_subparsers.add_parser(
         "list", help="List category topics by latest activity, optionally since a cutoff."
@@ -79,6 +85,7 @@ def _build_parser() -> argparse.ArgumentParser:
         "--limit",
         type=int,
         default=listing.DEFAULT_LIMIT,
-        help=f"Return at most this many unique topics (default {listing.DEFAULT_LIMIT}); look ahead to prove truncation.",
+        help=f"Return at most this many unique topics (default {listing.DEFAULT_LIMIT}); "
+        "look ahead to prove truncation. Shared read budgets still apply.",
     )
     return parser

tools/discourse/src/magpie_discourse/client.py

@@ -21,17 +21,21 @@
 from __future__ import annotations
 
 import http.client
+import io
 import json
 import os
 import re
 import urllib.error
 import urllib.parse
 import urllib.request
 from collections.abc import Mapping, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any
 
 DEFAULT_TIMEOUT_SECONDS = 30
+DEFAULT_MAX_REQUESTS = 64
+DEFAULT_MAX_BODY_BYTES = 32 * 1024 * 1024
+RESPONSE_CHUNK_BYTES = 64 * 1024
 
 QueryValue = str | int | Sequence[str | int]
 
@@ -40,14 +44,72 @@ class DiscourseError(Exception):
     """Raised when the Discourse bridge cannot complete a request."""
 
 
+class RequestBudgetError(DiscourseError):
+    """Raised when a read needs more request attempts than its budget allows."""
+
+
+@dataclass
+class ReadBudget:
+    """Mutable request and response-body accounting shared through DiscourseConfig."""
+
+    max_requests: int = DEFAULT_MAX_REQUESTS
+    max_body_bytes: int = DEFAULT_MAX_BODY_BYTES
+    requests: int = field(default=0, init=False)
+    body_bytes: int = field(default=0, init=False)
+
+    def start_request(self) -> None:
+        """Charge one permitted attempt before it reaches the transport."""
+        if self.requests >= self.max_requests:
+            raise RequestBudgetError(f"Discourse read budget exhausted: {self.max_requests} HTTP requests")
+        if self.body_bytes >= self.max_body_bytes:
+            raise DiscourseError(
+                f"Discourse read budget exhausted: {self.max_body_bytes} response-body bytes"
+            )
+        self.requests += 1
+
+    def read_body(self, response: Any) -> bytearray:
+        """Read within the remaining allowance, with one byte to detect overflow."""
+        body = bytearray()
+        while self.body_bytes <= self.max_body_bytes:
+            chunk = response.read(min(RESPONSE_CHUNK_BYTES, self.max_body_bytes - self.body_bytes + 1))
+            self.body_bytes += len(chunk)
+            if self.body_bytes > self.max_body_bytes:
+                break
+            if not chunk:
+                # HTTPResponse.read(size) can reach EOF without the IncompleteRead
+                # raised by read(); do not accept a truncated but valid JSON prefix.
+                unread = getattr(response, "length", None)
+                if isinstance(unread, int) and unread > 0:
+                    raise OSError("Discourse response ended before its declared Content-Length")
+                return body
+            body.extend(chunk)
+        raise DiscourseError(f"Discourse read budget exhausted: {self.max_body_bytes} response-body bytes")
+
+
 class SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler):
     """Keep read requests within the configured forum's origin and path."""
 
-    def __init__(self, base_url: str) -> None:
+    def __init__(self, base_url: str, budget: ReadBudget | None = None) -> None:
+        self._budget = budget if budget is not None else ReadBudget()
         self._origin = origin(parse_url(base_url))
         self._raw_path = urllib.parse.urlsplit(base_url).path.rstrip("/")
         self._path = _checked_path(base_url).rstrip("/")
 
+    def https_request(self, request: urllib.request.Request) -> urllib.request.Request:
+        """Apply the same check and budget to initial requests and every redirect."""
+        self.check_request(request)
+        self._budget.start_request()
+        return request
+
+    def http_error_302(self, req: urllib.request.Request, fp: Any, code: int, msg: str, headers: Any) -> Any:
+        """Keep urllib's redirect rules without draining an unused response body."""
+        if "location" not in headers and "uri" not in headers:
+            return None
+        fp.close()
+        return super().http_error_302(req, io.BytesIO(), code, msg, headers)
+
+    http_error_301 = http_error_303 = http_error_307 = http_error_308 = http_error_302
+
     def check_request(self, request: urllib.request.Request) -> None:
         """Reject writes and requests outside the configured forum before sending."""
         if request.get_method() != "GET":
@@ -134,15 +196,16 @@ def _is_under_path(path: str, base_path: str) -> bool:
 
 @dataclass(frozen=True)
 class DiscourseConfig:
-    """Environment-derived Discourse bridge configuration."""
+    """Frozen settings with a mutable read budget shared by operations on this instance."""
 
     base_url: str
     api_key: str | None
     api_username: str | None
+    budget: ReadBudget = field(default_factory=ReadBudget, repr=False, compare=False)
 
 
 def load_config() -> DiscourseConfig:
-    """Load the bridge configuration from environment variables."""
+    """Load environment settings with a fresh budget for one command."""
     raw_url = os.environ.get("DISCOURSE_URL", "").strip()
     if not raw_url:
         raise DiscourseError("DISCOURSE_URL is required (the forum's https:// base URL)")
@@ -195,9 +258,9 @@ def api_url(config: DiscourseConfig, path: str, params: Mapping[str, QueryValue]
     return url
 
 
-def _urlopen(request: urllib.request.Request, timeout: float, *, base_url: str) -> Any:
+def _urlopen(request: urllib.request.Request, timeout: float, *, base_url: str, budget: ReadBudget) -> Any:
     """Open a GET with the configured forum boundary (patched in tests)."""
-    handler = SameOriginRedirectHandler(base_url)
+    handler = SameOriginRedirectHandler(base_url, budget)
     handler.check_request(request)
     opener = urllib.request.build_opener(handler)
     return opener.open(request, timeout=timeout)
@@ -214,12 +277,15 @@ def get_json(
         method="GET",
     )
     try:
-        with _urlopen(request, DEFAULT_TIMEOUT_SECONDS, base_url=config.base_url) as response:
-            parsed = json.loads(response.read().decode("utf-8"))
+        with _urlopen(
+            request, DEFAULT_TIMEOUT_SECONDS, base_url=config.base_url, budget=config.budget
+        ) as response:
+            parsed = json.loads(config.budget.read_body(response).decode("utf-8"))
     except DiscourseError:
         raise
     except urllib.error.HTTPError as exc:
-        raise DiscourseError(_http_error_message(exc)) from exc
+        with exc:
+            raise DiscourseError(_http_error_message(exc, config.budget)) from exc
     except urllib.error.URLError as exc:
         raise DiscourseError(f"Failed to connect to Discourse: {exc.reason}") from exc
     except TimeoutError as exc:
@@ -233,15 +299,17 @@ def get_json(
     return parsed
 
 
-def _http_error_message(exc: urllib.error.HTTPError) -> str:
+def _http_error_message(exc: urllib.error.HTTPError, budget: ReadBudget) -> str:
     """Describe an HTTP error, keeping Discourse's own error text when it sends one."""
     message = f"Discourse request failed with HTTP {exc.code}"
     if exc.code == 429:
         retry_after = exc.headers.get("Retry-After") if exc.headers else None
         hint = f"{retry_after} seconds" if retry_after and retry_after.isdecimal() else retry_after
         return f"{message} (rate limited; retry after {hint or 'a while'})"
     try:
-        body = json.loads(exc.read().decode("utf-8"))
+        body = json.loads(budget.read_body(exc).decode("utf-8"))
+    except DiscourseError:
+        return f"{message} (error body exceeded the remaining read budget)"
     except (OSError, http.client.HTTPException, UnicodeDecodeError, json.JSONDecodeError):
         return message
     errors = body.get("errors") if isinstance(body, dict) else None

tools/discourse/src/magpie_discourse/topics.py

@@ -23,11 +23,18 @@
 import re
 from typing import Any
 
-from magpie_discourse.client import DiscourseConfig, DiscourseError, get_json, origin, parse_url
+from magpie_discourse.client import (
+    DiscourseConfig,
+    DiscourseError,
+    RequestBudgetError,
+    get_json,
+    origin,
+    parse_url,
+)
 
 #: Posts returned per ``/t/{id}/posts.json`` request; Discourse's own chunk size.
 POSTS_CHUNK_SIZE = 20
-#: Default ceiling on posts read for one topic; a longer topic is marked truncated.
+#: Default post-count limit; truncation is reported only after a successful read.
 DEFAULT_MAX_POSTS = 500
 
 _TOPIC_PATH = re.compile(r"^/t/(?:[^/]+/)?(?P<id>[0-9]+)(?:/[0-9]+)?(?:\.json)?/?$")
@@ -86,6 +93,13 @@ def fetch_topic(config: DiscourseConfig, topic_id: int, max_posts: int = DEFAULT
 
     wanted = post_ids[:max_posts]
     missing = [post_id for post_id in wanted if post_id not in posts]
+    requests_needed = (len(missing) + POSTS_CHUNK_SIZE - 1) // POSTS_CHUNK_SIZE
+    remaining_requests = config.budget.max_requests - config.budget.requests
+    if requests_needed > remaining_requests:
+        raise RequestBudgetError(
+            f"Discourse topic read for max_posts needs at least {requests_needed} more HTTP requests, "
+            f"but {remaining_requests} remain in the read budget"
+        )
     for start in range(0, len(missing), POSTS_CHUNK_SIZE):
         chunk = missing[start : start + POSTS_CHUNK_SIZE]
         extra = get_json(config, f"/t/{topic_id}/posts.json", {"post_ids[]": chunk})

tools/discourse/tests/test_discourse.py

@@ -54,7 +54,9 @@ def __init__(self, routes: dict[str, Any]) -> None:
         self.routes = routes
         self.requests: list[urllib.request.Request] = []
 
-    def __call__(self, request: urllib.request.Request, timeout: float, *, base_url: str) -> FakeResponse:
+    def __call__(
+        self, request: urllib.request.Request, timeout: float, *, base_url: str, budget: client.ReadBudget
+    ) -> FakeResponse:
         self.requests.append(request)
         path = urllib.parse.urlparse(request.full_url).path
         reply = self.routes[path]
@@ -314,7 +316,7 @@ def test_get_json_reports_rate_limiting(monkeypatch: pytest.MonkeyPatch, config:
 
 def test_get_json_rejects_non_json(monkeypatch: pytest.MonkeyPatch, config: DiscourseConfig) -> None:
     monkeypatch.setattr(
-        client, "_urlopen", lambda request, timeout, *, base_url: FakeResponse(b"<html>login</html>")
+        client, "_urlopen", lambda request, timeout, *, base_url, budget: FakeResponse(b"<html>login</html>")
     )
     with pytest.raises(DiscourseError, match="not JSON"):
         client.get_json(config, "/t/42.json")
@@ -562,7 +564,7 @@ def test_interrupted_http_error_body_keeps_the_http_status(
 ) -> None:
     error = _http_error(503)
 
-    def interrupted() -> bytes:
+    def interrupted(size: int = -1) -> bytes:
         raise http.client.IncompleteRead(b"partial", 10)
 
     monkeypatch.setattr(error, "read", interrupted)
@@ -763,7 +765,12 @@ def unexpected_open(*args: Any, **kwargs: Any) -> Any:
 
     monkeypatch.setattr(urllib.request, "build_opener", unexpected_open)
     with pytest.raises(DiscourseError):
-        client._urlopen(urllib.request.Request(url, method=method), 30, base_url=f"{BASE}/forum")
+        client._urlopen(
+            urllib.request.Request(url, method=method),
+            30,
+            base_url=f"{BASE}/forum",
+            budget=client.ReadBudget(),
+        )
 
 
 def test_cli_transport_binds_redirects_to_the_configured_forum(
@@ -975,7 +982,7 @@ def unexpected_open(*args: Any, **kwargs: Any) -> Any:
     monkeypatch.setattr(urllib.request, "build_opener", unexpected_open)
     request = urllib.request.Request(f"{BASE}{path}")
     with pytest.raises(DiscourseError):
-        client._urlopen(request, 30, base_url=f"{BASE}/forum")
+        client._urlopen(request, 30, base_url=f"{BASE}/forum", budget=client.ReadBudget())
 
 
 def _serve_post_variant(