Proxy meter: stop counting chrome:// and other local loads in Selenium and Puppeteer

CHANGELOG.md

@@ -60,6 +60,10 @@ that changes a default is called out at the top of its entry.
   `407` or `402` by name before any browser is started.
 - CI only: the canary now runs weekly (Mondays 06:41 UTC) and on a manual dispatch, with its optional secret `INSTAGRAM_PROXY` passed on every run; the daily schedule is removed.
 
+### Fixed
+
+- Proxy traffic meter: Selenium and Puppeteer no longer count requests that never leave the browser. Chrome's new-tab page loads `chrome://` scripts (about 2.8 MB, seen through a stub proxy) that were reported as finished loads and added to `proxy_bytes`, enough to bring a run noticeably closer to `--proxy-budget-mb`. Only `http(s)` requests and responses are counted now. The block sent once the budget is spent is also handed to the page's own event loop, so it works when the budget runs out on another thread.
+
 ## [0.3.0] - 2026-10-15
 
 > **Behaviour changes.** CSV output now prefixes a leading apostrophe to any

proxy_meter.py

@@ -58,7 +58,8 @@ def budget_mb(value: str) -> float:
     try:
         number = float(value)
     except ValueError:
-        raise argparse.ArgumentTypeError("--proxy-budget-mb must be a number of megabytes, got %r" % value)
+        raise argparse.ArgumentTypeError("--proxy-budget-mb must be a number of megabytes, got %r"
+                                         % value) from None
     if number != number or number < 0 or number == float("inf"):
         raise argparse.ArgumentTypeError("--proxy-budget-mb must be >= 0 (0 = unlimited), got %r" % value)
     return number
@@ -80,14 +81,15 @@ def _header_bytes(headers: Any) -> int:
     `[{name, value}]`)."""
     if not headers:
         return 0
+    pairs: Iterable[Any]
     if isinstance(headers, dict):
         pairs = headers.items()
     else:
         pairs = ((h.get("name", ""), h.get("value", "")) if isinstance(h, dict) else h for h in headers)
     return sum(len(str(k)) + len(str(v)) + 4 for k, v in pairs)
 
 
-def _request_line_bytes(method: str, url: str) -> int:
+def _request_line_bytes(method: Optional[str], url: str) -> int:
     parts = urlsplit(url or "")
     target = (parts.path or "/") + ("?" + parts.query if parts.query else "")
     return len("%s %s HTTP/1.1\r\n" % (method or "GET", target)) + 2  # + the blank line
@@ -103,8 +105,8 @@ def __init__(self, budget_mb: float = DEFAULT_BUDGET_MB, active: bool = False):
         self.exceeded = False
         self._lock = threading.Lock()
         self._reported = False
-        self._on_exceeded = []
-        self._partial = {}
+        self._on_exceeded: list = []
+        self._partial: dict = {}
 
     # ------------------------------------------------------------- counting
     def add(self, n: Any) -> bool:
@@ -247,19 +249,22 @@ def feed_cdp(self, method: str, params: dict) -> None:
                 self.add(redirect.get("encodedDataLength") or 0)
             request = params.get("request") or {}
             url = request.get("url") or ""
-            if url.startswith(("data:", "blob:", "about:", "chrome")):
-                return
+            if not url.startswith(("http://", "https://")):
+                return  # data:, blob:, chrome:// (the new-tab page), ... never cross the proxy
             body = request.get("postData") or ""
             self.add(_request_line_bytes(request.get("method"), url) + _header_bytes(request.get("headers"))
                      + len(body.encode("utf-8", "replace") if isinstance(body, str) else body))
         elif method == "Network.responseReceived":
             # Only a response that came over the network is counted at
             # loadingFinished: Chrome's own error page ("This site can't be
             # reached", ~250 KB) is reported as a finished load of the failed
-            # URL with no responseReceived, and a cache hit crossed nothing.
+            # URL with no responseReceived, the new-tab page's chrome://
+            # scripts (~2.8 MB, seen with Selenium) never leave the browser,
+            # and a cache hit crossed nothing.
             response = params.get("response") or {}
-            if not (response.get("fromDiskCache") or response.get("fromServiceWorker")
-                    or response.get("fromPrefetchCache")):
+            local = str(response.get("url") or "").split(":", 1)[0] not in ("", "http", "https")
+            if not local and not (response.get("fromDiskCache") or response.get("fromServiceWorker")
+                                  or response.get("fromPrefetchCache")):
                 with self._lock:
                     self._partial.setdefault(params.get("requestId"), 0)
         elif method == "Network.dataReceived":
@@ -300,8 +305,18 @@ async def attach_pyppeteer(self, page) -> None:
             client.on(event, lambda params, _event=event: self.feed_cdp(_event, params))
         await client.send("Network.enable", {})
         if self.budget_bytes:
+            # The budget can trip on another thread (a concurrent worker) that
+            # has no event loop: hand the block to this page's own loop.
+            loop = asyncio.get_event_loop()
+
+            async def _send_block():
+                try:
+                    await client.send("Network.setBlockedURLs", {"urls": ["*"]})
+                except Exception:  # noqa: BLE001 - the loop check still stops the run
+                    pass
+
             def _block():
-                asyncio.ensure_future(client.send("Network.setBlockedURLs", {"urls": ["*"]}))
+                loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_send_block()))
             self.on_exceeded(_block)
 
     def attach_selenium(self, driver, *, keep_entries: bool = True) -> None:

tests/test_proxy_meter.py

@@ -224,7 +224,13 @@ def test_loading_finished_and_request_sizes(self):
         meter.feed_cdp("Network.loadingFinished", {"requestId": "e", "encodedDataLength": 250609})
         meter.feed_cdp("Network.responseReceived", {"requestId": "c", "response": {"fromDiskCache": True}})
         meter.feed_cdp("Network.loadingFinished", {"requestId": "c", "encodedDataLength": 900})
-        self.assertEqual(meter.bytes, sent + 1300, "an error page or a cache hit crossed no proxy")
+        meter.feed_cdp("Network.requestWillBeSent", {"requestId": "n", "request": {
+            "method": "GET", "url": "chrome://new-tab-page/x.js", "headers": {"A": "b"}}})
+        meter.feed_cdp("Network.responseReceived", {"requestId": "n",
+                                                    "response": {"url": "chrome://new-tab-page/x.js"}})
+        meter.feed_cdp("Network.loadingFinished", {"requestId": "n", "encodedDataLength": 6575})
+        self.assertEqual(meter.bytes, sent + 1300,
+                         "an error page, a cache hit or a chrome:// page crossed no proxy")
         before = meter.bytes
         meter.feed_cdp("Network.requestWillBeSent", {"requestId": "3", "request": {"method": "GET", "url": "data:x"}})
         self.assertEqual(meter.bytes, before, "a data: URL never crosses the proxy")
@@ -298,6 +304,36 @@ async def go():
         self.assertEqual(meter.bytes, 2000)
         self.assertIn(("Network.setBlockedURLs", {"urls": ["*"]}), client.sent)
 
+    def test_pyppeteer_block_from_a_worker_thread(self):
+        """The budget can trip on a thread with no event loop (a concurrent
+        worker): the block is still sent, on the page's own loop."""
+        sent = []
+
+        class Client:
+            def on(self, name, fn):
+                pass
+
+            async def send(self, cmd, params=None):
+                sent.append(cmd)
+
+        async def create():
+            return Client()
+
+        page = argparse.Namespace(target=argparse.Namespace(createCDPSession=create))
+        meter = proxy_meter.ProxyMeter(0.001, active=True)
+
+        async def go():
+            await meter.attach_pyppeteer(page)
+            worker = threading.Thread(target=meter.add, args=(5000,))
+            worker.start()
+            worker.join()
+            for _ in range(5):
+                await asyncio.sleep(0)
+
+        asyncio.run(go())
+        self.assertTrue(meter.exceeded)
+        self.assertIn("Network.setBlockedURLs", sent)
+
 
 class PlaywrightHooks(unittest.TestCase):
     def test_sizes_and_budget_route(self):