feat: Add large command chunking

README.md

@@ -101,34 +101,34 @@ See [the devtools reference][devtools-ref] for a list of possible commands.
 Try adding the following to the example shown above:
 
 ```python
-    # Callback for printing result
-    async def dump_event(response):
-        print(str(response))
-
-
-    # Callback for raising result as error
-    async def error_event(response):
-        raise Exception(str(response))
-
-
-    browser.subscribe("Target.targetCrashed", error_event)
-    new_tab.subscribe("Page.loadEventFired", dump_event)
-    browser.subscribe("Target.*", dump_event) # dumps all "Target" events
-    response = await new_tab.subscribe_once("Page.lifecycleEvent")
-    # do something with response
-    browser.unsubscribe("Target.*")
-    # events are always sent to a browser or tab,
-    # but the documentation isn't always clear which.
-    # Dumping all: `browser.subscribe("*", dump_event)` (on tab too)
-    # can be useful (but verbose) for debugging.
+# Callback for printing result
+async def dump_event(response):
+    print(str(response))
+
+
+# Callback for raising result as error
+async def error_event(response):
+    raise Exception(str(response))
+
+
+browser.subscribe("Target.targetCrashed", error_event)
+new_tab.subscribe("Page.loadEventFired", dump_event)
+browser.subscribe("Target.*", dump_event)  # dumps all "Target" events
+response = await new_tab.subscribe_once("Page.lifecycleEvent")
+# do something with response
+browser.unsubscribe("Target.*")
+# events are always sent to a browser or tab,
+# but the documentation isn't always clear which.
+# Dumping all: `browser.subscribe("*", dump_event)` (on tab too)
+# can be useful (but verbose) for debugging.
 ```
 
 ## Synchronous Use
 
 You can use this library without `asyncio`,
 
 ```python
-my_browser = choreo.Browser() # blocking until open
+my_browser = choreo.Browser()  # blocking until open
 ```
 
 However, you must call `browser.pipe.read_jsons(blocking=True|False)` manually,

pyproject.toml

@@ -81,21 +81,22 @@ src = ["src"]
 select = ["ALL"]
 ignore = [
           "ANN", # no types
-          "EM", # allow strings in raise(), despite python being ugly about it
-          "TRY003", # allow long error messages inside raise()
+          "COM812", # manual says linter rule conflicts with formatter
+          "CPY001", # Don't require a copyright notice at the top of a file
           "D203", # No blank before class docstring (D211 = require blank line)
           "D212", # Commit message style docstring is D213, ignore D212
-          "COM812", # manual says linter rule conflicts with formatter
+          "EM", # allow strings in raise(), despite python being ugly about it
+          "G004", # fstrings in my logs
           "ISC001", # manual says litner rule conflicts with formatter
+          "PT003", # scope="function" implied but I like readability
           "RET504", # Allow else if unnecessary because more readable
           "RET505", # Allow else if unnecessary because more readable
           "RET506", # Allow else if unnecessary because more readable
           "RET507", # Allow else if unnecessary because more readable
           "RET508", # Allow else if unnecessary because more readable
           "RUF012", # We don't do typing, so no typing
           "SIM105", # Too opionated (try-except-pass)
-          "PT003", # scope="function" implied but I like readability
-          "G004", # fstrings in my logs
+          "TRY003", # allow long error messages inside raise()
           ]
 
 [tool.ruff.lint.per-file-ignores]
@@ -112,14 +113,20 @@ asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 log_cli = false
 addopts = "--import-mode=append"
+markers = [
+  "slow: moves enough data to be worth skipping by default (`-m 'not slow'`)",
+]
 
 # tell poe to use the env we give it, otherwise it detects uv and overrides flags
 [tool.poe]
 executor.type = "simple"
 
 [tool.poe.tasks]
 test_proc = "pytest --log-level=1 -W error -n auto -v -rfE --capture=fd tests/test_process.py"
-test_fn = "pytest --log-level=1 -W error -n auto -v -rfE --capture=fd --ignore=tests/test_process.py"
+test_fn = "pytest --log-level=1 -W error -n auto -v -rfE --capture=fd -m 'not slow' --ignore=tests/test_process.py"
+# Skip `--log-level=1` and `-n auto` to allow slow tests to complete without
+# formatting wall of text and to avoid tests using up all memory (~1GB per thread)
+test_slow = "pytest -W error -v -rfE --capture=fd -m slow --ignore=tests/test_process.py"
 debug-test_proc = "pytest --log-level=1 -W error -vvvx -rA --show-capture=no --capture=no tests/test_process.py"
 debug-test_fn = "pytest --log-level=1 -W error -vvvx -rA --show-capture=no --capture=no --ignore=tests/test_process.py"
 

src/choreographer/channels/__init__.py

@@ -5,14 +5,20 @@
 
 """
 
-from ._errors import BlockWarning, ChannelClosedError, JSONError
+from ._errors import (
+    BlockWarning,
+    ChannelClosedError,
+    JSONError,
+    MessageTooLargeError,
+)
 from ._wire import register_custom_encoder
 from .pipe import Pipe
 
 __all__ = [
     "BlockWarning",
     "ChannelClosedError",
     "JSONError",
+    "MessageTooLargeError",
     "Pipe",
     "register_custom_encoder",
 ]

src/choreographer/channels/_errors.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+
 class BlockWarning(UserWarning):
     """A warning for when block modification operations used on incompatible OS."""
 
@@ -8,3 +11,43 @@ class ChannelClosedError(IOError):
 
 class JSONError(RuntimeError):
     """Another JSONError."""
+
+
+class MessageTooLargeError(RuntimeError):
+    """
+    An error for when a message won't fit in the browser's receive buffer.
+
+    The browser closes the connection outright if we write a message bigger
+    than its buffer, so we refuse to write it and raise this instead.
+    """
+
+    size: int
+    """The size, in bytes, of the message we refused to write."""
+    max_size: int
+    """The largest message the browser will accept."""
+    payload: str | None
+    """
+    The serialized message.
+
+    It is kept so that callers who know how to break the message up can
+    reuse it instead of serializing all over again. It is deliberately left
+    out of the error text: it can be hundreds of megabytes.
+    """
+
+    def __init__(self, size: int, max_size: int, payload: str | None = None) -> None:
+        """
+        Construct a MessageTooLargeError.
+
+        Args:
+            size: the size of the message in bytes.
+            max_size: the largest message the browser will accept.
+            payload: the serialized message, if the caller should have it.
+
+        """
+        super().__init__(
+            f"Message is {size} bytes, which is over the browser's "
+            f"{max_size} byte limit. It was not sent.",
+        )
+        self.size = size
+        self.max_size = max_size
+        self.payload = payload

src/choreographer/channels/_wire.py

@@ -41,7 +41,17 @@ def default(self, o: Any) -> Any:
         return simplejson.JSONEncoder.default(self, o)
 
 
-def serialize(obj: Any) -> bytes:
+def serialize_str(obj: Any) -> str:
+    """
+    Serialize an object to a JSON string.
+
+    Use `serialize()` to return a value that's ready to send to Chrome.
+    This exists for callers that need to slice the string up first.
+
+    Args:
+        obj: Any Python object that serializes to JSON.
+
+    """
     try:
         if not _custom_encoder:
             message = simplejson.dumps(
@@ -57,7 +67,11 @@ def serialize(obj: Any) -> bytes:
     _logger.debug(f"Serialized: {message[:15]}...{message[-15:]}, size: {len(message)}")
     _logger.debug2(f"Whole message: {message}")
 
-    return message.encode("utf-8")
+    return message
+
+
+def serialize(obj: Any) -> bytes:
+    return serialize_str(obj).encode("utf-8")
 
 
 def deserialize(message: str) -> Any:

src/choreographer/channels/pipe.py

@@ -13,7 +13,12 @@
 import logistro
 
 from . import _wire as wire
-from ._errors import BlockWarning, ChannelClosedError, JSONError
+from ._errors import (
+    BlockWarning,
+    ChannelClosedError,
+    JSONError,
+    MessageTooLargeError,
+)
 
 if TYPE_CHECKING:
     from typing import Any, Mapping, Sequence
@@ -24,6 +29,14 @@
 
 _logger = logistro.getLogger(__name__)
 
+MAX_MESSAGE_SIZE = 100 * 1024 * 1024
+"""
+The biggest message Chrome will read off the pipe, in bytes.
+
+This mirrors `kReceiveBufferSizeForDevTools` in Chrome's
+`content/browser/devtools/devtools_pipe_handler.cc`.
+"""
+
 # should be closing my ends from the start?
 
 
@@ -84,15 +97,36 @@ def write_json(self, obj: Mapping[str, Any]) -> tuple[float, float]:
         Send one json down the pipe.
 
         Args:
-            obj: any python object that serializes to json.
+            obj: Any python object that serializes to JSON.
+
+        Raises:
+            ChannelClosedError: If the pipe was never opened or is already
+                closed, or if the OS write fails. A failed write closes the
+                pipe, so nothing can be sent after this.
+            MessageTooLargeError: If the message won't fit in Chrome's buffer.
+                Nothing is written, so the channel is still good afterwards.
+                The error carries the serialized message so that callers who
+                can break it up don't have to serialize it a second time.
+            TypeError: If `obj` contains something the encoder doesn't know how
+                to turn into JSON.
+            UnicodeEncodeError: If the serialized message contains lone
+                surrogates, which have no UTF-8 representation.
 
         """
         if not self.is_ready():
             raise ChannelClosedError(
                 "The communication channel was either never "
                 "opened or closed. Was .open() or .close() called?",
             )
-        encoded_message = wire.serialize(obj) + b"\0"
+        message = wire.serialize_str(obj)
+        encoded_message = message.encode("utf-8") + b"\0"
+        if len(encoded_message) > MAX_MESSAGE_SIZE:
+            # Don't close(): we haven't written anything, the pipe is fine.
+            raise MessageTooLargeError(
+                len(encoded_message),
+                MAX_MESSAGE_SIZE,
+                payload=message,
+            )
         _logger.debug(
             f"Writing message {encoded_message[:15]!r}...{encoded_message[-15:]!r}, "
             f"size: {len(encoded_message)}.",

src/choreographer/cli/_cli_utils.py

@@ -131,7 +131,7 @@ def get_chrome_sync(  # noqa: C901, PLR0912, PLR0915
 
     if i:
         _logger.info("Loading chrome from list")
-        raw_json = urllib.request.urlopen(  # noqa: S310 audit url for schemes
+        raw_json = urllib.request.urlopen(
             _chrome_for_testing_url,
         ).read()
         browser_list = json.loads(