From 610d7b4806d031fed782945d02c08ce670993773 Mon Sep 17 00:00:00 2001
From: jehrr <79864894+jehrr@users.noreply.github.com>
Date: Thu, 22 Oct 2026 22:26:55 +0000
Subject: [PATCH] Retry a throttled page turn once after a 30-60 s pause;
 canary walk budget 15 MB

A press for the next page that gets no grid back (the site's /graphql
answered 429 with no Retry-After) is now retried once in all three
engines, after a random pause of 2/3 to 4/3 of --throttle-wait-seconds
(default 45, 0 = off), in the same browser session and exit. The proxy
budget is checked before the pause and before the second press. A second
failure ends the run as before: partial, exit 6, next_page_throttled.
The listing sidecar records throttle_retries.

The canary's three-page walk passes --proxy-budget-mb 15 (the scraper's
default stays 10): with --block-media the 2026-10-22 walk used 8.7 MB for
pages 1 and 2 at 50 of 50 cards each.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CKsd7q25N19WzUZjrAdzx8
---
 .github/proxy_traffic.py     |  14 +-
 .github/workflows/canary.yml |  14 +-
 CHANGELOG.md                 |  34 +++-
 README.md                    |  24 ++-
 TROUBLESHOOTING.md           |  12 +-
 page_flow.py                 | 114 +++++++++++-
 playwright_scraper.py        |  29 ++-
 puppeteer_scraper.py         |  31 +++-
 selenium_scraper.py          |  40 +++-
 smoke_test.py                | 346 ++++++++++++++++++++++++++++++++++-
 10 files changed, 610 insertions(+), 48 deletions(-)

diff --git a/.github/proxy_traffic.py b/.github/proxy_traffic.py
index a0ad740..2330de6 100644
--- a/.github/proxy_traffic.py
+++ b/.github/proxy_traffic.py
@@ -5,12 +5,16 @@
 Reads `<out>.latest_attempt.meta.json` (written on every metered run, rows
 or not), else `<out>.meta.json`, and appends
 
-    **<label>**: proxy traffic X.X MB of 10 MB
+    **<label>**: proxy traffic X.X MB of N MB
 
-to $GITHUB_STEP_SUMMARY. A missing or unreadable sidecar, or a run that did
-not go through a local proxy, is reported as such and is not a failure here
-(the exit-code steps judge the run itself). A run that went over its proxy
-budget prints `::error::stopped: proxy budget 10 MB exceeded` and exits 1.
+to $GITHUB_STEP_SUMMARY. N is the run's OWN budget, read from the sidecar's
+`proxy_budget_bytes` (written by the scraper from --proxy-budget-mb; `null`
+or absent reads as "no budget"), so a canary step that raises its budget is
+reported against that budget rather than a fixed figure. A missing or
+unreadable sidecar, or a run that did not go through a local proxy, is
+reported as such and is not a failure here (the exit-code steps judge the
+run itself). A run that went over its proxy budget prints
+`::error::stopped: proxy budget N MB exceeded` and exits 1.
 """
 
 import json
diff --git a/.github/workflows/canary.yml b/.github/workflows/canary.yml
index e6a01be..7fce826 100644
--- a/.github/workflows/canary.yml
+++ b/.github/workflows/canary.yml
@@ -163,20 +163,28 @@ jobs:
         # 2026-10-21 walk used 9.6 MB of the 10 MB cap with two of its three
         # pages only 19 cards deep, so three full pages with images would
         # not fit. image_url is not one of the columns this job checks.
+        # --proxy-budget-mb 15, not the scraper's default 10: with
+        # --block-media the 2026-10-22 walk (run 39858684619) used 8.7 MB
+        # for pages 1 and 2 at 50 of 50 cards each, so a full third page and
+        # one paced retry of a throttled turn would not fit under 10.
+        # A turn that gets no grid back (a /graphql 429) is retried ONCE
+        # after a 30-60 s pause (--throttle-wait-seconds, default 45);
+        # throttle_retries in the sidecar says whether that happened.
         run: |
           set +e
           start=$(date -u -d '+45 days' +%F)
           end=$(date -u -d '+47 days' +%F)
           url="https://www.vrbo.com/search?destination=Orlando%2C%20Florida%2C%20United%20States%20of%20America&regionId=2693&sort=RECOMMENDED&startDate=${start}&endDate=${end}&adults=2"
           xvfb-run -a python3 playwright_scraper.py --url "$url" \
-            --pages 3 --delay 15 --retries 4 --proxy-budget-mb 10 --block-media \
+            --pages 3 --delay 15 --retries 4 --proxy-budget-mb 15 --block-media \
             --out canary_walk --dump-html canary_walk.html
           echo "exit_code=$?" >> "$GITHUB_OUTPUT"
 
       - name: Proxy traffic of the walk
         # The walk goes through VRBO_PROXY, which meters what it sends and
-        # stops at --proxy-budget-mb 10. This puts the figure in the step
-        # summary and fails the job when the cap was hit.
+        # stops at --proxy-budget-mb 15. This puts the figure in the step
+        # summary, with the budget read from the run's own sidecar, and fails
+        # the job when the cap was hit.
         if: ${{ !cancelled() && steps.run.outcome != 'skipped' }}
         run: python3 .github/proxy_traffic.py "pagination" canary_walk
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3ec84b2..e55e88e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,14 @@ that is a fix, and the release notes lead with it.
 
 ## [Unreleased]
 
+> **Behaviour change.** A page turn that gets no grid back (the site's
+> `/graphql` answered HTTP 429, with no `Retry-After`) is now retried
+> **once** in all three engines, after a random 30-60 s pause, in the same
+> browser session and exit. A run that used to end `next_page_throttled`
+> (exit 6) at once can now take up to about two minutes longer at that
+> page, and goes on to the next page when the retry succeeds.
+> `--throttle-wait-seconds 0` restores the old behaviour.
+
 > **Behaviour change.** On a 2Captcha residential gateway
 > (`*.proxy.2captcha.com`, `*.proxy.rucaptcha.com`), a proxy login with no
 > `-session-` segment now gets one random `-session-<id>-sessTime-30`
@@ -34,6 +42,18 @@ that is a fix, and the release notes lead with it.
 
 ### Added
 
+- `--throttle-wait-seconds` in all three engines (default 45, `0` = no
+  retry): one paced retry of a page turn that got no grid back. The pause is
+  a random 2/3 to 4/3 of the value. The proxy budget is checked before the
+  pause and again before the second press, so a spent budget never buys a
+  retry. A grid that arrives during the pause counts as the turn and is not
+  pressed again. The listing sidecar gains `throttle_retries` (retries made
+  in the run). Measured on the 2026-10-22 canary walk (run 39858684619):
+  pages 1 and 2 turned in 2 s, then the press for page 3 got one `/graphql`
+  429 with no `Retry-After`, and the grid stayed empty for 60 s. Offline
+  checks drive each engine's real page fetch with a stubbed grid: 429 then
+  success reads the page, 429 twice ends exit 6 after exactly two presses,
+  and a budget spent by the failed turn makes no second press.
 - `--proxy-session-minutes` in all three engines, and
   `VRBO_PROXY_SESSION_MINUTES` in `.env.example`. The session id is drawn
   once per run, before the pre-check; with `--proxy-file`, each gateway line
@@ -55,9 +75,10 @@ that is a fix, and the release notes lead with it.
     `proxy_budget_exceeded`, and a closing log line
     `Proxy traffic: X.X MB of Y MB budget`.
   - `--proxy-budget-mb`, and `VRBO_PROXY_BUDGET_MB` (in `.env.example`).
-- CI only: both canary runs pass `--proxy-budget-mb 10`. Each one writes its
-  proxy traffic to the step summary, and fails with
-  `::error::stopped: proxy budget 10 MB exceeded` when the cap was hit
+- CI only: both canary runs pass a `--proxy-budget-mb` (10 for the first
+  page, 15 for the walk; see Changed). Each one writes its proxy traffic to
+  the step summary, and fails with
+  `::error::stopped: proxy budget N MB exceeded` when the cap was hit
   (`.github/proxy_traffic.py`).
 - Offline checks:
   - The meter against known CDP and Playwright sizes.
@@ -88,6 +109,13 @@ that is a fix, and the release notes lead with it.
 
 ### Changed
 
+- CI only: the canary's three-page walk through `VRBO_PROXY` passes
+  `--proxy-budget-mb 15`; the scraper's default stays 10. With
+  `--block-media` the 2026-10-22 walk used 8.7 MB for pages 1 and 2 at 50 of
+  50 cards each, so a full third page and a paced retry would not fit under
+  10. The step summary already reported each run against the budget in its
+  own sidecar (`proxy_budget_bytes`); its docstring said "of 10 MB" and now
+  says so, and the canary checks cover a 15 MB run and a run with no cap.
 - CI only: the whole canary, free and paid jobs alike, now runs weekly (Mondays 06:20 UTC) and on a manual dispatch; the daily schedule is removed.
 
 ### Fixed
diff --git a/README.md b/README.md
index 22a55a1..407259a 100644
--- a/README.md
+++ b/README.md
@@ -127,10 +127,17 @@ while every `/graphql` POST behind a press came back **HTTP 429**
 (`{"error":"Too Many Requests","message":"Provisioned request rate has been
 exceeded"}`) and the site's own client gave up.
 
-When that happens the run reports **`partial`, exit 6** — never "the listing
-ended" — and says how many 429s it counted. `--delay` (default **5.0s**,
-higher than the family default) is the cheap lever; `--proxy-file` is the one
-that scales.
+The 429 carries no `Retry-After` header. So when a press gets no grid back,
+the run waits a random **30-60 s** (`--throttle-wait-seconds`, default **45**,
+`0` = off) in the same browser session and exit and presses next **once**
+more. The next page exists only inside that session, so it is never retried
+from a new browser or a new exit. The proxy budget is checked before the
+retry. The sidecar records `throttle_retries`.
+
+If the retry gets no grid either, the run reports **`partial`, exit 6** —
+never "the listing ended" — and says how many 429s it counted (Playwright
+only). `--delay` (default **5.0s**, higher than the family default) is the
+cheap lever; `--proxy-file` is the one that scales.
 
 ### 4. The results list is an inner scroller, and the page body never scrolls
 
@@ -260,6 +267,10 @@ expedia_property_id, price_note, sku, url, source. The only differences were
 Engine flag differences are asserted in both directions by the test suite, so
 closing one needs a README edit rather than a quiet patch:
 
+* `--throttle-wait-seconds SECONDS` — all three engines, but not part of the
+  family flag contract. The pause before the one retry of a page turn that
+  got no grid back: a random 2/3 to 4/3 of it (30-60 s at the default 45).
+  `0` turns the retry off.
 * `--block-retry-delay SECONDS` — all three engines, but not part of the
   family flag contract. Wait before re-fetching a page that came back
   refused, doubling per retry (30 waits 30 s, then 60 s). Unset keeps the
@@ -315,7 +326,10 @@ A proxy is billed by traffic. With `--proxy` / `--proxy-file` / `VRBO_PROXY`:
   Media is loaded by default here, and images are most of a search page's
   bytes, so a long walk through a proxy reaches 10 MB sooner than on a site
   that blocks them. Raise the budget, or pass `--block-media` if `image_url`
-  is not needed. The sidecars record `proxy_bytes`, `proxy_budget_bytes`
+  is not needed. Measured on the canary's walk through a residential exit
+  (2026-10-22, run 39858684619): with `--block-media`, pages 1 and 2 at 50 of
+  50 cards each used **8.7 MB**, so that walk runs at `--proxy-budget-mb 15`
+  to leave room for its third page and a paced retry. The sidecars record `proxy_bytes`, `proxy_budget_bytes`
   (`null` with no cap) and `proxy_budget_exceeded`: in `<out>.meta.json`,
   and in `<out>.latest_attempt.meta.json`, which every metered run writes.
   The log ends with `Proxy traffic: X.X MB of Y MB budget`.
diff --git a/TROUBLESHOOTING.md b/TROUBLESHOOTING.md
index ab9e0fc..a4ac181 100644
--- a/TROUBLESHOOTING.md
+++ b/TROUBLESHOOTING.md
@@ -50,12 +50,20 @@ Look at `stop_reason` in `<out>.meta.json`.
 on the page. The data behind a next-press comes over `POST /graphql`, which
 has its own limiter: the HTML keeps answering 200 while the turn is refused
 with `{"error":"Too Many Requests","message":"Provisioned request rate has
-been exceeded"}`. The log says how many 429s it counted.
+been exceeded"}`, with no `Retry-After` header. The log says how many 429s
+it counted (Playwright only; Selenium and pyppeteer cannot see the response).
+
+Before it stops, the run has already retried that turn once: it waited a
+random 30-60 s (`--throttle-wait-seconds`, default 45) in the same browser
+session and exit, then pressed next again. `throttle_retries` in the sidecar
+counts those retries. `next_page_throttled` means the retry was refused too
+(or the proxy budget was spent first, which reads `proxy_budget_exceeded`).
 
 In order of cost:
 
 1. `--delay 15` (the default is 5)
-2. wait a few minutes — this clears on its own
+2. a longer `--throttle-wait-seconds`, or wait a few minutes and re-run —
+   this clears on its own
 3. `--proxy-file` to spread the load across exits
 
 It is reported as partial rather than complete **on purpose**. Reporting a
diff --git a/page_flow.py b/page_flow.py
index a9a1f2d..1b8860c 100644
--- a/page_flow.py
+++ b/page_flow.py
@@ -53,7 +53,8 @@
 from __future__ import annotations
 
 import logging
-from typing import Callable, Dict, List, Optional
+import random
+from typing import Callable, Dict, List, Optional, Tuple
 
 from product_parser import (SELECTORS, NEXT_PAGE_SELECTOR, PAGE_CAP,
                             challenge_vendor, detect_block_marker,
@@ -635,25 +636,122 @@ def advance_to_next_page(press_next: Callable[[], bool],
     way, and on this site the reason is almost always that the `/graphql`
     POST behind the turn was answered 429.
     """
-    before = first_card_href()
-    before_sku = sku_from_url(strip_tracking(before or "")) if before else None
+    before_sku = _first_sku(first_card_href)
     if not press_next():
         return NO_BUTTON
+    return _await_turnover(first_card_href, count, sleep, before_sku,
+                           timeout_ms)
+
+
+def _first_sku(first_card_href: Callable[[], Optional[str]]) -> Optional[str]:
+    href = first_card_href()
+    return sku_from_url(strip_tracking(href or "")) if href else None
+
+
+def _turned_over(first_card_href, count, before_sku) -> bool:
+    if count(READY_SELECTOR_LISTING) <= 0:
+        return False
+    now_sku = _first_sku(first_card_href)
+    return bool(now_sku and now_sku != before_sku)
+
+
+def _await_turnover(first_card_href, count, sleep, before_sku,
+                    timeout_ms: int) -> str:
     waited = 0
     while waited < timeout_ms:
         sleep(NEXT_PAGE_POLL_MS)
         waited += NEXT_PAGE_POLL_MS
-        if count(READY_SELECTOR_LISTING) <= 0:
-            continue
-        now = first_card_href()
-        now_sku = sku_from_url(strip_tracking(now or "")) if now else None
-        if now_sku and now_sku != before_sku:
+        if _turned_over(first_card_href, count, before_sku):
             logger.info("next page arrived after %dms", waited)
             return ADVANCED
     logger.info("next page never turned over within %dms", timeout_ms)
     return NO_TURNOVER
 
 
+# One bounded retry of a page turn that never came back.
+#
+# Measured on the 2026-10-22 canary walk through a sticky residential
+# session: pages 1 and 2 turned over in 2 s, then the press for page 3 got
+# one `/graphql` HTTP 429 ("Provisioned request rate has been exceeded",
+# no Retry-After header) and the grid stayed empty for the whole 60 s. The
+# answer carries no wait to honour, so the wait is ours: a random pause
+# between 2/3 and 4/3 of `--throttle-wait-seconds` (30-60 s at the default
+# 45), in the SAME browser, session and exit — the next page exists only
+# inside this session, so a relaunch or a new exit would restart the listing
+# at page 1. Then the turn is tried ONCE more. A second failure ends the run
+# as before (partial, next_page_throttled); there is no third press.
+#
+# The retry keys on the no-turnover STATE rather than on a counted 429: the
+# Playwright engine can count the /graphql 429s, Selenium has no response
+# listener and cannot. All three engines therefore retry the same state and
+# report the count only where it is known.
+THROTTLE_WAIT_DEFAULT_S = 45.0
+THROTTLE_RETRIES_PER_TURN = 1
+
+
+def throttle_pause_s(wait_seconds: float,
+                     uniform: Callable[[float, float], float] = random.uniform
+                     ) -> float:
+    """The randomised pause before retrying a throttled turn, in seconds."""
+    return uniform(wait_seconds * 2.0 / 3.0, wait_seconds * 4.0 / 3.0)
+
+
+def advance_with_throttle_retry(press_next: Callable[[], bool],
+                                first_card_href: Callable[[], Optional[str]],
+                                count: Callable[[str], int],
+                                sleep: Callable[[int], None],
+                                wait_seconds: float,
+                                before_retry: Callable[[], None],
+                                throttled: Callable[[], int] = lambda: 0,
+                                uniform: Callable[[float, float], float]
+                                = random.uniform,
+                                timeout_ms: int = NEXT_PAGE_TIMEOUT_MS
+                                ) -> Tuple[str, int]:
+    """`advance_to_next_page`, with one paced retry of a turn that failed.
+
+    Returns (outcome, retries made). `wait_seconds` 0 disables the retry.
+    `before_retry` is called before the pause and again before the second
+    press; the engines pass the proxy-budget check, which raises once the
+    run's budget is spent, so a spent budget never buys a retry.
+
+    The second attempt compares against the card that was first BEFORE the
+    original press: a grid that arrived late, during the pause, is a turn
+    and is not pressed again (a second press would skip a page). A next
+    button that is gone after the failed turn is NOT the end of the
+    listing here — it reads as another failed turn, never as NO_BUTTON.
+    """
+    before_sku = _first_sku(first_card_href)
+    if not press_next():
+        return NO_BUTTON, 0
+    turn = _await_turnover(first_card_href, count, sleep, before_sku,
+                           timeout_ms)
+    if turn == ADVANCED or not wait_seconds or wait_seconds <= 0:
+        return turn, 0
+    before_retry()
+    pause = throttle_pause_s(wait_seconds, uniform)
+    seen = throttled()
+    logger.warning(
+        "The page turn did not come back within %.0fs%s. Waiting %.0fs in "
+        "the same browser session and exit, then pressing next once more "
+        "(--throttle-wait-seconds %g; 0 disables this retry).",
+        timeout_ms / 1000,
+        f" ({seen} /graphql response(s) were HTTP 429)" if seen else "",
+        pause, wait_seconds)
+    sleep(int(pause * 1000))
+    before_retry()
+    if _turned_over(first_card_href, count, before_sku):
+        logger.info("The next page arrived during the pause; not pressing "
+                    "again.")
+        return ADVANCED, THROTTLE_RETRIES_PER_TURN
+    if not press_next():
+        logger.info("The next button is not on the page after the failed "
+                    "turn; the turn stays failed.")
+        return NO_TURNOVER, THROTTLE_RETRIES_PER_TURN
+    turn = _await_turnover(first_card_href, count, sleep, before_sku,
+                           timeout_ms)
+    return turn, THROTTLE_RETRIES_PER_TURN
+
+
 def page_cap_reached(page_num: int) -> bool:
     return page_num >= PAGE_CAP
 
diff --git a/playwright_scraper.py b/playwright_scraper.py
index 85966e0..0e1845e 100644
--- a/playwright_scraper.py
+++ b/playwright_scraper.py
@@ -181,6 +181,8 @@ class PageOutcome:
     # page_flow.unobtained_reason). Not ok, so it is never "the catalogue is
     # empty".
     unobtained: Optional[str] = None
+    # Paced retries of a page turn that never came back (0 or 1).
+    throttle_retries: int = 0
 
     @property
     def ok(self) -> bool:
@@ -803,11 +805,16 @@ def _fetch_one_page(session, args, pool, page_num: int, url: Optional[str]) -> P
                         page_num, args.pages, page_num)
             proxy_meter.check(args)
             throttled_before = _graphql_throttled(session)
-            turn = page_flow.advance_to_next_page(
+            # One paced retry of a turn that never came back, in this same
+            # browser session (see page_flow.advance_with_throttle_retry).
+            turn, outcome.throttle_retries = page_flow.advance_with_throttle_retry(
                 lambda: _press_next(session.page),
                 lambda: _first_card_href(session.page),
                 lambda sel: _count(session.page, sel),
-                session.page.wait_for_timeout)
+                lambda ms: session.page.wait_for_timeout(ms),
+                args.throttle_wait_seconds,
+                lambda: proxy_meter.check(args),
+                throttled=lambda: _graphql_throttled(session) - throttled_before)
             if turn == page_flow.NO_BUTTON:
                 # The listing genuinely ran out. A complete answer.
                 outcome.state = "exhausted"
@@ -827,7 +834,7 @@ def _fetch_one_page(session, args, pool, page_num: int, url: Optional[str]) -> P
                 outcome.final_url = session.page.url
                 logger.error(
                     "Pressed next for page %d and the grid never came back "
-                    "within %.0fs%s. This is NOT the end of the listing — "
+                    "within %.0fs%s%s. This is NOT the end of the listing — "
                     "the run is reported as PARTIAL (exit 6) rather than "
                     "complete. The data behind a page turn is fetched over "
                     "POST /graphql, which is rate limited SEPARATELY from the "
@@ -835,6 +842,7 @@ def _fetch_one_page(session, args, pool, page_num: int, url: Optional[str]) -> P
                     "is refused. Raise --delay (it is %.1fs now), or spread "
                     "the load with --proxy-file.",
                     page_num, page_flow.NEXT_PAGE_TIMEOUT_MS / 1000,
+                    " on both attempts" if outcome.throttle_retries else "",
                     f" — {throttled} /graphql response(s) were HTTP 429"
                     if throttled else "", args.delay)
                 return outcome
@@ -1099,6 +1107,7 @@ def _fetch_one_page(session, args, pool, page_num: int, url: Optional[str]) -> P
 def scrape(args) -> int:
     outcomes: List[PageOutcome] = []
     seen_keys = set()
+    throttle_retries = 0
     blocked = False
     dedupe_key = "sku"
     stop_reason = "single_page_mode" if args.mode == "property" else "completed"
@@ -1174,6 +1183,7 @@ def scrape(args) -> int:
 
                     time.sleep(args.delay)
                     outcome = _fetch_one_page(session, args, pool, page_num, None)
+                    throttle_retries += outcome.throttle_retries
                     outcomes.append(outcome)
 
                     if outcome.state == "exhausted":
@@ -1301,7 +1311,8 @@ def scrape(args) -> int:
                  "cards_missing": cards_missing,
                  "results_total": total_available,
                  "results_total_is_floor": total_is_floor,
-                 "pages_still_growing": unsettled}
+                 "pages_still_growing": unsettled,
+                 "throttle_retries": throttle_retries}
         if unsettled:
             logger.warning(
                 "Page(s) %s were still filling when the scroll budget ran "
@@ -1389,6 +1400,16 @@ def parse_args():
                         "measured clearing minutes later, so a longer wait "
                         "is the lever that fits it. The number of retries "
                         "stays the built-in budget (2 without a proxy pool).")
+    p.add_argument("--throttle-wait-seconds", type=page_flow.at_least(0, float),
+                   default=page_flow.THROTTLE_WAIT_DEFAULT_S, metavar="SECONDS",
+                   help="When the press for the next page gets no new grid "
+                        "(the site's /graphql answered HTTP 429, with no "
+                        "Retry-After), wait a random 2/3 to 4/3 of this "
+                        "(30-60 s at the default 45) in the same browser "
+                        "session and exit, and press once more. One retry "
+                        "per turn; a second failure ends the run as partial "
+                        "(exit 6). The proxy budget is checked before the "
+                        "retry. 0 = no retry.")
     p.add_argument("--format", choices=["json", "csv", "both"], default="both")
     p.add_argument("--out", default="vrbo_products", help="Output file prefix")
     p.add_argument("--locale", default="en-US",
diff --git a/puppeteer_scraper.py b/puppeteer_scraper.py
index 8f7fe48..1a3da0d 100644
--- a/puppeteer_scraper.py
+++ b/puppeteer_scraper.py
@@ -199,6 +199,8 @@ class PageOutcome:
     scroll: Optional[dict] = None
     # Served but not obtained; see playwright_scraper.PageOutcome.
     unobtained: Optional[str] = None
+    # Paced retries of a page turn that never came back (0 or 1).
+    throttle_retries: int = 0
 
     @property
     def ok(self) -> bool:
@@ -582,11 +584,15 @@ def _fetch_one_page(session, args, pool, page_num: int,
                         "next button (this listing has no page-%d address).",
                         page_num, args.pages, page_num)
             proxy_meter.check(args)
-            turn = page_flow.advance_to_next_page(
+            # One paced retry of a turn that never came back, in this same
+            # browser session (see page_flow.advance_with_throttle_retry).
+            turn, outcome.throttle_retries = page_flow.advance_with_throttle_retry(
                 lambda: _press_next(session),
                 lambda: _first_card_href(session),
                 lambda sel: _count(session, sel),
-                _sleep)
+                lambda ms: _sleep(ms),
+                args.throttle_wait_seconds,
+                lambda: proxy_meter.check(args))
             if turn == page_flow.NO_BUTTON:
                 outcome.state = "exhausted"
                 outcome.final_url = _current_url(session)
@@ -601,12 +607,14 @@ def _fetch_one_page(session, args, pool, page_num: int,
                 outcome.final_url = _current_url(session)
                 logger.error(
                     "Pressed next for page %d and the grid never came back "
-                    "within %.0fs. This is NOT the end of the listing — the "
+                    "within %.0fs%s. This is NOT the end of the listing — the "
                     "run is reported as PARTIAL (exit 6). The data behind a "
                     "page turn is fetched over POST /graphql, which is rate "
                     "limited SEPARATELY from the HTML. Raise --delay (it is "
                     "%.1fs now), or spread the load with --proxy-file.",
-                    page_num, page_flow.NEXT_PAGE_TIMEOUT_MS / 1000, args.delay)
+                    page_num, page_flow.NEXT_PAGE_TIMEOUT_MS / 1000,
+                    " on both attempts" if outcome.throttle_retries else "",
+                    args.delay)
                 return outcome
 
         if load_failed:
@@ -802,6 +810,7 @@ def _fetch_one_page(session, args, pool, page_num: int,
 def scrape(args) -> int:
     outcomes: List[PageOutcome] = []
     seen_keys = set()
+    throttle_retries = 0
     blocked = False
     stop_reason = "single_page_mode" if args.mode == "property" else "completed"
 
@@ -851,6 +860,7 @@ def scrape(args) -> int:
                         "session.", page_num)
                 time.sleep(args.delay)
                 outcome = _fetch_one_page(session, args, pool, page_num, None)
+                throttle_retries += outcome.throttle_retries
                 outcomes.append(outcome)
 
                 if outcome.state == "exhausted":
@@ -946,7 +956,8 @@ def scrape(args) -> int:
                  "cards_missing": cards_missing,
                  "results_total": total_available,
                  "results_total_is_floor": total_is_floor,
-                 "pages_still_growing": unsettled}
+                 "pages_still_growing": unsettled,
+                 "throttle_retries": throttle_retries}
 
     extra = session_fields(pool, extra)
     extra = proxy_meter.finish(args, extra)
@@ -996,6 +1007,16 @@ def parse_args():
                         "measured clearing minutes later, so a longer wait "
                         "is the lever that fits it. The number of retries "
                         "stays the built-in budget (2 without a proxy pool).")
+    p.add_argument("--throttle-wait-seconds", type=page_flow.at_least(0, float),
+                   default=page_flow.THROTTLE_WAIT_DEFAULT_S, metavar="SECONDS",
+                   help="When the press for the next page gets no new grid "
+                        "(the site's /graphql answered HTTP 429, with no "
+                        "Retry-After), wait a random 2/3 to 4/3 of this "
+                        "(30-60 s at the default 45) in the same browser "
+                        "session and exit, and press once more. One retry "
+                        "per turn; a second failure ends the run as partial "
+                        "(exit 6). The proxy budget is checked before the "
+                        "retry. 0 = no retry.")
     p.add_argument("--format", choices=["json", "csv", "both"], default="both")
     p.add_argument("--out", default="vrbo_products", help="Output file prefix")
     p.add_argument("--chromium-path", default=None, metavar="PATH",
diff --git a/selenium_scraper.py b/selenium_scraper.py
index 0580fb5..e35e610 100644
--- a/selenium_scraper.py
+++ b/selenium_scraper.py
@@ -144,6 +144,8 @@ class PageOutcome:
     scroll: Optional[dict] = None
     # Served but not obtained; see playwright_scraper.PageOutcome.
     unobtained: Optional[str] = None
+    # Paced retries of a page turn that never came back (0 or 1).
+    throttle_retries: int = 0
 
     @property
     def ok(self) -> bool:
@@ -379,6 +381,15 @@ def _press_next(session) -> bool:
         return False
 
 
+def _budget_check(session, args) -> None:
+    """Count what the browser has sent so far, then enforce the budget.
+
+    Selenium's traffic is read from the performance log, so it has to be
+    drained before the check means anything."""
+    proxy_meter.drain_selenium(session.driver)
+    proxy_meter.check(args)
+
+
 def _graphql_throttled(session) -> int:
     """How many /graphql responses came back 429.
 
@@ -570,11 +581,15 @@ def _fetch_one_page(session, args, pool, page_num: int,
                         page_num, args.pages, page_num)
             proxy_meter.drain_selenium(session.driver)
             proxy_meter.check(args)
-            turn = page_flow.advance_to_next_page(
+            # One paced retry of a turn that never came back, in this same
+            # browser session (see page_flow.advance_with_throttle_retry).
+            turn, outcome.throttle_retries = page_flow.advance_with_throttle_retry(
                 lambda: _press_next(session),
                 lambda: _first_card_href(session),
                 lambda sel: _count(session, sel),
-                _sleep)
+                lambda ms: _sleep(ms),
+                args.throttle_wait_seconds,
+                lambda: _budget_check(session, args))
             if turn == page_flow.NO_BUTTON:
                 outcome.state = "exhausted"
                 outcome.final_url = _current_url(session)
@@ -590,14 +605,16 @@ def _fetch_one_page(session, args, pool, page_num: int,
                 outcome.final_url = _current_url(session)
                 logger.error(
                     "Pressed next for page %d and the grid never came back "
-                    "within %.0fs. This is NOT the end of the listing — the "
+                    "within %.0fs%s. This is NOT the end of the listing — the "
                     "run is reported as PARTIAL (exit 6). The data behind a "
                     "page turn is fetched over POST /graphql, which is rate "
                     "limited SEPARATELY from the HTML: the page itself still "
                     "answers 200 while the turn is refused. Raise --delay "
                     "(it is %.1fs now), or spread the load with "
                     "--proxy-file.",
-                    page_num, page_flow.NEXT_PAGE_TIMEOUT_MS / 1000, args.delay)
+                    page_num, page_flow.NEXT_PAGE_TIMEOUT_MS / 1000,
+                    " on both attempts" if outcome.throttle_retries else "",
+                    args.delay)
                 return outcome
 
         if load_failed:
@@ -790,6 +807,7 @@ def _fetch_one_page(session, args, pool, page_num: int,
 def scrape(args) -> int:
     outcomes: List[PageOutcome] = []
     seen_keys = set()
+    throttle_retries = 0
     blocked = False
     stop_reason = "single_page_mode" if args.mode == "property" else "completed"
 
@@ -839,6 +857,7 @@ def scrape(args) -> int:
                         "restart the listing at page 1.", page_num)
                 time.sleep(args.delay)
                 outcome = _fetch_one_page(session, args, pool, page_num, None)
+                throttle_retries += outcome.throttle_retries
                 outcomes.append(outcome)
 
                 if outcome.state == "exhausted":
@@ -933,7 +952,8 @@ def scrape(args) -> int:
                  "cards_missing": cards_missing,
                  "results_total": total_available,
                  "results_total_is_floor": total_is_floor,
-                 "pages_still_growing": unsettled}
+                 "pages_still_growing": unsettled,
+                 "throttle_retries": throttle_retries}
 
     # Selenium strips a proxy login (see _Session), so no session is sent.
     extra = session_fields(pool, extra, login_sent=False)
@@ -990,6 +1010,16 @@ def parse_args():
                         "measured clearing minutes later, so a longer wait "
                         "is the lever that fits it. The number of retries "
                         "stays the built-in budget (2 without a proxy pool).")
+    p.add_argument("--throttle-wait-seconds", type=page_flow.at_least(0, float),
+                   default=page_flow.THROTTLE_WAIT_DEFAULT_S, metavar="SECONDS",
+                   help="When the press for the next page gets no new grid "
+                        "(the site's /graphql answered HTTP 429, with no "
+                        "Retry-After), wait a random 2/3 to 4/3 of this "
+                        "(30-60 s at the default 45) in the same browser "
+                        "session and exit, and press once more. One retry "
+                        "per turn; a second failure ends the run as partial "
+                        "(exit 6). The proxy budget is checked before the "
+                        "retry. 0 = no retry.")
     p.add_argument("--format", choices=["json", "csv", "both"], default="both")
     p.add_argument("--out", default="vrbo_products", help="Output file prefix")
     p.add_argument("--locale", default="en-US",
diff --git a/smoke_test.py b/smoke_test.py
index 40831c5..6dd3855 100644
--- a/smoke_test.py
+++ b/smoke_test.py
@@ -50,7 +50,7 @@
 import re
 import sys
 import tempfile
-from dataclasses import asdict, fields
+from dataclasses import asdict, fields, replace
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
@@ -1444,11 +1444,12 @@ def test_engine_parity(skips):
         "playwright_scraper": {"--locale", "--fingerprint", "--fp-tags",
                                "--fp-country", "--browser-channel",
                                "--cdp-connect-timeout",
-                               "--block-retry-delay"},
+                               "--block-retry-delay", "--throttle-wait-seconds"},
         "selenium_scraper": {"--locale", "--fingerprint", "--fp-tags",
-                             "--fp-country", "--block-retry-delay"},
+                             "--fp-country", "--block-retry-delay",
+                             "--throttle-wait-seconds"},
         "puppeteer_scraper": {"--chromium-path", "--cdp-connect-timeout",
-                              "--block-retry-delay"},
+                              "--block-retry-delay", "--throttle-wait-seconds"},
     }
     for engine, extra in documented_extra.items():
         actual = flagsets[engine] - contract
@@ -2255,7 +2256,8 @@ def test_numeric_flags_are_validated():
             ("--pages", int, 1, "0", "2"), ("--concurrency", int, 1, "0", "1"),
             ("--proxy-block-retries", int, 0, "-1", "0"),
             ("--delay", float, 0, "-1", "0"), ("--retry-delay", float, 0, "-0.5", "1.5"),
-            ("--block-retry-delay", float, 0, "-30", "30")):
+            ("--block-retry-delay", float, 0, "-30", "30"),
+            ("--throttle-wait-seconds", float, 0, "-1", "0")):
         parse = page_flow.at_least(minimum, kind)
         ok &= check(f"{flag}: {good} is accepted", parse(good) == kind(good))
         ok &= check(f"{flag}: {bad} is refused",
@@ -2267,6 +2269,7 @@ def test_numeric_flags_are_validated():
              '"--concurrency", type=page_flow.at_least(1)',
              '"--retry-delay", type=page_flow.at_least(0, float)',
              '"--block-retry-delay", type=page_flow.at_least(0, float)',
+             '"--throttle-wait-seconds", type=page_flow.at_least(0, float)',
              '"--proxy-block-retries", type=page_flow.at_least(0)')
     for engine in ENGINES:
         src = open(os.path.join(REPO_ROOT, f"{engine}.py"), encoding="utf-8").read()
@@ -3663,9 +3666,26 @@ def test_canary_reports_proxy_traffic():
     steps = re.split(r"\n\s*- (?:name|uses):", text)
     runs = [s for s in steps if re.search(r"\w+_scraper\.py", s)
             and "--out" in s]
-    ok &= check(f"every canary scrape passes --proxy-budget-mb 10 "
-                f"({len(runs)} run step(s))", bool(runs) and all(
-                    "--proxy-budget-mb 10" in r for r in runs))
+    budgets = {}
+    for r in runs:
+        code = "\n".join(line for line in r.splitlines()
+                         if not line.lstrip().startswith("#"))
+        for out_ in re.findall(r"--out\s+(\S+)", code):
+            budgets[out_] = re.findall(r"--proxy-budget-mb\s+(\S+)", code)
+    # The walk carries three full pages and a paced retry of a throttled
+    # turn: 8.7 MB for two pages measured 2026-10-22 with --block-media, so
+    # 15 there. The first page has no proxy and keeps the default 10.
+    ok &= check(f"every canary scrape passes one --proxy-budget-mb: the "
+                f"first page 10, the walk 15 ({budgets})",
+                budgets == {"canary_run": ["10"], "canary_walk": ["15"]})
+    walk = [r for r in runs if "canary_walk" in r]
+    ok &= check("the walk does not switch the throttle retry off",
+                bool(walk) and "--throttle-wait-seconds 0" not in walk[0])
+    ok &= check("no summary step passes a fixed budget",
+                all(re.search(r"proxy_traffic\.py\s+\S+\s+\S+\s*\n", s + "\n")
+                    and not re.search(r"proxy_traffic\.py\s+\S+\s+\S+\s+\d",
+                                      s)
+                    for s in steps if ".github/proxy_traffic.py" in s))
     outs = set()
     for r in runs:
         outs.update(re.findall(r"--out\s+(\S+)", r))
@@ -3689,6 +3709,15 @@ def test_canary_reports_proxy_traffic():
                 ("under", {"proxy_bytes": 2_340_000,
                            "proxy_budget_bytes": 10_000_000,
                            "proxy_budget_exceeded": False}),
+                ("walk15", {"proxy_bytes": 8_666_843,
+                            "proxy_budget_bytes": 15_000_000,
+                            "proxy_budget_exceeded": False}),
+                ("over15", {"proxy_bytes": 15_100_000,
+                            "proxy_budget_bytes": 15_000_000,
+                            "proxy_budget_exceeded": True}),
+                ("nocap", {"proxy_bytes": 1_000_000,
+                           "proxy_budget_bytes": None,
+                           "proxy_budget_exceeded": False}),
                 ("plain", {"status": "complete"}),
                 ("missing", None)):
             out = os.path.join(tmp, case)
@@ -3710,12 +3739,312 @@ def test_canary_reports_proxy_traffic():
                 results["under"][0] == 0
                 and "**live/under**: proxy traffic 2.3 MB of 10 MB" in written
                 and "**live/over**: proxy traffic 12.3 MB of 10 MB" in written)
+    ok &= check("a 15 MB run is reported against ITS budget: '**live/walk15**: "
+                "proxy traffic 8.7 MB of 15 MB'",
+                results["walk15"][0] == 0
+                and "**live/walk15**: proxy traffic 8.7 MB of 15 MB" in written)
+    ok &= check("over a 15 MB budget: '::error::stopped: proxy budget 15 MB "
+                "exceeded' and the step fails",
+                results["over15"][0] == 1 and "::error::stopped: proxy budget "
+                "15 MB exceeded" in results["over15"][1])
+    ok &= check("a null proxy_budget_bytes reads as no cap",
+                results["nocap"][0] == 0
+                and "**live/nocap**: proxy traffic 1.0 MB of no budget" in written)
     ok &= check("no proxy fields / no sidecar: reported, not a crash",
                 results["plain"][0] == 0 and results["missing"][0] == 0
                 and "not metered" in written and "unknown" in written)
     return ok
 
 
+class _TurnSite:
+    """A results grid whose next-press turns over only on chosen presses.
+
+    `turn_on` holds the press numbers (1-based) whose /graphql answer
+    arrives; every other press is the measured throttle: the press goes
+    through and the first card never changes. `on_press` runs on each press
+    (a test uses it to spend the proxy budget)."""
+
+    def __init__(self, turn_on=(), button=True, late=False, on_press=None):
+        self.turn_on, self.button, self.late = set(turn_on), button, late
+        self.on_press = on_press
+        self.presses, self.sku, self.sleeps = 0, 1, []
+
+    def press(self, *a):
+        if self.presses and not self.button:
+            return False
+        self.presses += 1
+        if self.on_press:
+            self.on_press()
+        if self.presses in self.turn_on:
+            self.sku = 100 + self.presses
+        return True
+
+    def href(self, *a):
+        return f"/pdp/lo/{self.sku}?pwa_ts=1"
+
+    def count(self, *a):
+        return 50
+
+    def sleep(self, ms):
+        self.sleeps.append(ms)
+        # The grid that answers late: it arrives during the paced pause.
+        if self.late and ms >= 1000 * 10 and self.presses == 1:
+            self.sku = 999
+
+    @property
+    def long_waits(self):
+        return [ms for ms in self.sleeps if ms > page_flow.NEXT_PAGE_POLL_MS]
+
+
+def _turn(site, wait=45.0, before_retry=lambda: None, throttled=lambda: 1):
+    return page_flow.advance_with_throttle_retry(
+        site.press, site.href, site.count, site.sleep, wait, before_retry,
+        throttled=throttled, timeout_ms=5_000)
+
+
+class _RecordingPool(_FakePool):
+    def __init__(self):
+        self.advances = 0
+
+    def advance(self, reason=""):
+        self.advances += 1
+
+
+def _run_fake_turn(engine_mod, site, wait=45.0, meter=None, tmp="."):
+    """Turn to page 3 through the engine's real `_fetch_one_page` (url=None),
+    with the press, the first card, the count and every sleep stubbed."""
+    import types
+    import proxy_meter
+    row = rows_of("LISTING_US")[0]
+    patches = {
+        "_press_next": site.press,
+        "_first_card_href": site.href,
+        "_count": site.count,
+        "_sleep": site.sleep,
+        "_graphql_throttled": lambda *a: 0,
+        "_content": lambda *a, **k: "<html></html>",
+        "_content_when_settled": lambda *a, **k: "<html></html>",
+        "_current_url": lambda *a, **k: _SAVERS_URL,
+        "_scroll_the_grid": lambda *a, **k: {},
+        "_parse_for_mode": lambda *a, **k: [row],
+        "detect_recaptcha_v3": lambda *a, **k: None,
+        "detect_recaptcha_in_page": lambda *a, **k: None,
+        "reconcile_detections": lambda *a, **k: None,
+        "time": types.SimpleNamespace(sleep=lambda s: site.sleeps.append(
+            int(s * 1000))),
+    }
+    saved = {k: getattr(engine_mod, k) for k in patches if hasattr(engine_mod, k)}
+    saved_classify = page_flow.classify
+    for k, v in patches.items():
+        if k in saved:
+            setattr(engine_mod, k, v)
+    page_flow.classify = lambda *a, **k: "content"
+    args = types.SimpleNamespace(
+        pages=3, retries=1, retry_delay=0, block_retry_delay=None,
+        proxy_block_retries=2, mode="listing", out=os.path.join(tmp, "turn"),
+        dump_html=None, solve_captcha="when-blocked",
+        twocaptcha_key=None, captcha_api="v2", min_score=0.7,
+        headless=False, delay=0, chromium_path="/stub", cdp_endpoint=None,
+        throttle_wait_seconds=wait)
+    if meter is not None:
+        setattr(args, proxy_meter._ARGS_ATTR, meter)
+    session, pool = _FakeSession(200), _RecordingPool()
+    session.page.wait_for_timeout = site.sleep
+    try:
+        outcome = engine_mod._fetch_one_page(session, args, pool, 3, None)
+        err = None
+    except BaseException as e:  # noqa: BLE001 — the budget stop is one
+        outcome, err = None, e
+    finally:
+        for k, v in saved.items():
+            setattr(engine_mod, k, v)
+        page_flow.classify = saved_classify
+    return outcome, err, session.relaunches + pool.advances
+
+
+def _scrape_with_page3(engine_mod, tmp, page3):
+    """`scrape()` over 3 pages with the page fetch faked: pages 1-2 serve
+    rows, page 3 is whatever `page3(PageOutcome)` makes of it."""
+    import contextlib as _ctx
+    from unittest import mock
+    rows = _meter_rows()
+    name = engine_mod.__name__
+    if "playwright" in name:
+        sessions = ["sync_playwright", "_BrowserSession"]
+    elif "puppeteer" in name:
+        sessions = ["_AsyncBridge", "_Session"]
+    else:
+        sessions = ["_Session"]
+
+    def fake_fetch(session, a, pool, page_num, url):
+        page_rows = [replace(r, sku=f"{page_num}{i}")
+                     for i, r in enumerate(rows)]
+        oc = engine_mod.PageOutcome(page_num=page_num, url=url or _SAVERS_URL,
+                                    final_url=_SAVERS_URL, products=page_rows,
+                                    state="content")
+        return page3(oc) if page_num == 3 else oc
+
+    args = _engine_args(engine_mod, tmp, None, "--pages", "3", "--delay", "0",
+                        "--format", "json")
+    with _ctx.ExitStack() as stack:
+        for attr in sessions:
+            stack.enter_context(mock.patch.object(engine_mod, attr))
+        stack.enter_context(mock.patch.object(engine_mod, "_fetch_one_page",
+                                              side_effect=fake_fetch))
+        stack.enter_context(mock.patch.object(engine_mod.time, "sleep"))
+        stack.enter_context(_captured_output())
+        try:
+            rc = engine_mod.scrape(args)
+        except BaseException as e:  # noqa: BLE001 — fail BY NAME
+            rc = f"raised {type(e).__name__}"
+    out = os.path.join(tmp, "out")
+    meta = (json.load(open(out + ".meta.json"))
+            if os.path.exists(out + ".meta.json") else {})
+    return rc, meta
+
+
+def test_throttle_retry():
+    group("A throttled page turn gets ONE paced retry in the same session")
+    import proxy_meter
+    ok = True
+
+    site = _TurnSite(turn_on={1})
+    ok &= check("a turn that comes back is not retried or paused",
+                _turn(site) == (page_flow.ADVANCED, 0) and site.presses == 1
+                and not site.long_waits)
+    site, calls = _TurnSite(turn_on={2}), []
+    got = _turn(site, before_retry=lambda: calls.append(1))
+    ok &= check(f"429 then success: the second press turns the page "
+                f"({got}, presses {site.presses})",
+                got == (page_flow.ADVANCED, 1) and site.presses == 2)
+    ok &= check(f"...after ONE pause of 30-60 s at the default 45 "
+                f"({site.long_waits} ms)",
+                len(site.long_waits) == 1
+                and 30_000 <= site.long_waits[0] <= 60_000)
+    ok &= check("...and the proxy budget is checked before the pause and "
+                "before the second press", len(calls) == 2)
+    site = _TurnSite()
+    got = _turn(site)
+    ok &= check(f"429 twice: the turn stays failed after exactly two presses "
+                f"({got}, presses {site.presses})",
+                got == (page_flow.NO_TURNOVER, 1) and site.presses == 2
+                and len(site.long_waits) == 1)
+    site = _TurnSite()
+
+    def spent():
+        raise proxy_meter.ProxyBudgetExceeded("proxy budget 0.0 MB exceeded")
+    err = _base_exception(lambda: _turn(site, before_retry=spent))
+    ok &= check(f"budget spent: no pause and no second press, the budget "
+                f"stop propagates ({type(err).__name__})",
+                isinstance(err, proxy_meter.ProxyBudgetExceeded)
+                and site.presses == 1 and not site.long_waits)
+    site = _TurnSite()
+    ok &= check("--throttle-wait-seconds 0: no retry",
+                _turn(site, wait=0) == (page_flow.NO_TURNOVER, 0)
+                and site.presses == 1 and not site.long_waits)
+    site = _TurnSite(late=True)
+    ok &= check("a grid that arrives DURING the pause is a turn, and is not "
+                "pressed again (a second press would skip a page)",
+                _turn(site) == (page_flow.ADVANCED, 1) and site.presses == 1)
+    site = _TurnSite(button=False)
+    ok &= check("a next button gone after the failed turn is still a failed "
+                "turn, never the end of the listing",
+                _turn(site) == (page_flow.NO_TURNOVER, 1))
+    site = _TurnSite(button=False)
+    site.presses = 0
+    site.button = True
+    site.press = lambda *a: False
+    ok &= check("no next button on the first press is still NO_BUTTON",
+                _turn(site) == (page_flow.NO_BUTTON, 0))
+    pauses = [page_flow.throttle_pause_s(45.0) for _ in range(300)]
+    ok &= check(f"the pause is random within 30-60 s at 45 "
+                f"({min(pauses):.1f}-{max(pauses):.1f})",
+                all(30.0 <= p <= 60.0 for p in pauses)
+                and len({round(p, 3) for p in pauses}) > 1
+                and page_flow.throttle_pause_s(45.0, lambda a, b: a) == 30.0
+                and page_flow.throttle_pause_s(45.0, lambda a, b: b) == 60.0)
+    ok &= check("the default is 45 s and one retry per turn",
+                page_flow.THROTTLE_WAIT_DEFAULT_S == 45.0
+                and page_flow.THROTTLE_RETRIES_PER_TURN == 1)
+
+    group("...through each engine's real page fetch")
+    with tempfile.TemporaryDirectory() as tmp:
+        for engine in ENGINES:
+            mod = _load_engine_copy(engine)
+            short = engine.split("_")[0]
+            site = _TurnSite(turn_on={2})
+            oc, err, moved = _run_fake_turn(mod, site, tmp=tmp)
+            ok &= check(f"{short}: 429 then success → page 3 read in the same "
+                        f"session (presses {site.presses}, "
+                        f"retries {getattr(oc, 'throttle_retries', None)}, "
+                        f"{err!r})",
+                        err is None and oc.ok and len(oc.products) == 1
+                        and oc.throttle_retries == 1 and site.presses == 2
+                        and moved == 0)
+            ok &= check(f"{short}: ...after one 30-60 s pause "
+                        f"({site.long_waits})",
+                        len([w for w in site.long_waits if w >= 30_000]) == 1
+                        and all(w <= 60_000 for w in site.long_waits))
+            site = _TurnSite()
+            oc, err, moved = _run_fake_turn(mod, site, tmp=tmp)
+            ok &= check(f"{short}: 429 twice → no_turnover after two presses, "
+                        f"throttle_retries 1, no relaunch or new exit",
+                        err is None and oc.state == "no_turnover"
+                        and oc.load_failed and oc.throttle_retries == 1
+                        and site.presses == 2 and moved == 0)
+            meter = proxy_meter.ProxyMeter(0.001)
+            site = _TurnSite(on_press=lambda: meter.add(5_000))
+            oc, err, moved = _run_fake_turn(mod, site, meter=meter, tmp=tmp)
+            ok &= check(f"{short}: budget spent by the failed turn → no retry "
+                        f"({type(err).__name__}, presses {site.presses})",
+                        isinstance(err, proxy_meter.ProxyBudgetExceeded)
+                        and site.presses == 1
+                        and not [w for w in site.long_waits if w >= 30_000])
+            site = _TurnSite(turn_on={2})
+            oc, err, moved = _run_fake_turn(mod, site, wait=3.0, tmp=tmp)
+            ok &= check(f"{short}: --throttle-wait-seconds reaches the turn "
+                        f"(3 → {site.long_waits})",
+                        oc is not None and oc.ok
+                        and [w for w in site.long_waits if w >= 2_000] != []
+                        and all(w <= 4_000 for w in site.long_waits))
+            site = _TurnSite(turn_on={2})
+            oc, err, moved = _run_fake_turn(mod, site, wait=0, tmp=tmp)
+            ok &= check(f"{short}: --throttle-wait-seconds 0 → no retry",
+                        oc is not None and oc.state == "no_turnover"
+                        and oc.throttle_retries == 0 and site.presses == 1)
+            args = _engine_args(mod, tmp, None)
+            ok &= check(f"{short}: --throttle-wait-seconds defaults to 45",
+                        args.throttle_wait_seconds == 45.0)
+
+            group(f"{short}: the run's exit and sidecar")
+
+            def retried_ok(oc):
+                oc.throttle_retries = 1
+                return oc
+
+            def retried_failed(oc):
+                oc.throttle_retries, oc.products = 1, []
+                oc.state, oc.load_failed = "no_turnover", True
+                return oc
+            rc, meta = _scrape_with_page3(mod, tmp, retried_ok)
+            ok &= check(f"{short}: a turn that succeeded on its retry → exit 0, "
+                        f"3 pages, throttle_retries 1 (rc={rc})",
+                        rc == 0 and meta.get("pages_completed") == 3
+                        and meta.get("throttle_retries") == 1
+                        and meta.get("status") == "complete")
+            rc, meta = _scrape_with_page3(mod, tmp, retried_failed)
+            ok &= check(f"{short}: throttled twice → exit 6 next_page_throttled, "
+                        f"throttle_retries 1 (rc={rc}, "
+                        f"{meta.get('stop_reason')})",
+                        rc == 6 and meta.get("stop_reason") == "next_page_throttled"
+                        and meta.get("throttle_retries") == 1
+                        and meta.get("pages_completed") == 2)
+            rc, meta = _scrape_with_page3(mod, tmp, lambda oc: oc)
+            ok &= check(f"{short}: no throttle → throttle_retries 0 in the "
+                        f"sidecar", rc == 0 and meta.get("throttle_retries") == 0)
+    return ok
+
+
 def main() -> int:
     ok = True
     skips = []
@@ -3732,6 +4061,7 @@ def main() -> int:
     ok &= test_scroll_loop()
     ok &= test_grid_lazy_tail()
     ok &= test_throttle_is_not_completion()
+    ok &= test_throttle_retry()
     ok &= test_output_contract()
     ok &= test_writers_and_finish_run()
     ok &= test_diff()