⚡ Bolt: Pre-compile regular expressions

.jules/bolt.md

@@ -0,0 +1,3 @@
+## 2024-05-01 - [Python Regex Precompilation]
+**Learning:** Python code in this application often performs repeated regex matches on incoming request headers or payload lines (e.g. `domain_fronter.py` content-range matching or cookie splitting).
+**Action:** Extract inline `re.search` or `re.split` inside high-traffic loops into module-level variables like `RE_CONTENT_RANGE = re.compile(...)`. This yields measurable CPU and memory savings by skipping implicit dictionary lookups and potential cache evictions in the internal Python re cache.

src/domain_fronter.py

@@ -32,6 +32,12 @@
 
 log = logging.getLogger("Fronter")
 
+# Pre-compiled regexes for performance
+RE_CONTENT_RANGE = re.compile(r"/(\d+)")
+RE_JSON_EXTRACT = re.compile(r'\{.*\}', re.DOTALL)
+RE_STATUS_CODE = re.compile(r"\d{3}")
+RE_COOKIE_SPLIT = re.compile(r",\s*(?=[A-Za-z0-9!#$%&'*+\-.^_`|~]+=)")
+
 
 class DomainFronter:
     def __init__(self, config: dict):
@@ -590,7 +596,7 @@ async def relay_parallel(self, method: str, url: str,
 
         # Parse total size from Content-Range: "bytes 0-262143/1048576"
         content_range = resp_hdrs.get("content-range", "")
-        m = re.search(r"/(\d+)", content_range)
+        m = RE_CONTENT_RANGE.search(content_range)
         if not m:
             # Can't parse — downgrade to 200 so the client (which sent a
             # plain GET) doesn't get confused by 206 + Content-Range.
@@ -1012,7 +1018,7 @@ def _parse_batch_body(self, resp_body: bytes,
         try:
             data = json.loads(text)
         except json.JSONDecodeError:
-            m = re.search(r'\{.*\}', text, re.DOTALL)
+            m = RE_JSON_EXTRACT.search(text)
             data = json.loads(m.group()) if m else None
         if not data:
             raise RuntimeError(f"Bad batch response: {text[:200]}")
@@ -1049,7 +1055,7 @@ async def _read_http_response(self, reader: asyncio.StreamReader):
         lines = header_section.split(b"\r\n")
 
         status_line = lines[0].decode(errors="replace")
-        m = re.search(r"\d{3}", status_line)
+        m = RE_STATUS_CODE.search(status_line)
         status = int(m.group()) if m else 0
 
         headers = {}
@@ -1139,7 +1145,7 @@ def _parse_relay_response(self, body: bytes) -> bytes:
         try:
             data = json.loads(text)
         except json.JSONDecodeError:
-            m = re.search(r'\{.*\}', text, re.DOTALL)
+            m = RE_JSON_EXTRACT.search(text)
             if m:
                 try:
                     data = json.loads(m.group())
@@ -1201,7 +1207,7 @@ def _split_set_cookie(blob: str) -> list[str]:
             return []
         # Split on ", " but only when the following text looks like the start
         # of a new cookie (a token followed by '=').
-        parts = re.split(r",\s*(?=[A-Za-z0-9!#$%&'*+\-.^_`|~]+=)", blob)
+        parts = RE_COOKIE_SPLIT.split(blob)
         return [p.strip() for p in parts if p.strip()]
 
     def _split_raw_response(self, raw: bytes):
@@ -1210,7 +1216,7 @@ def _split_raw_response(self, raw: bytes):
             return 0, {}, raw
         header_section, body = raw.split(b"\r\n\r\n", 1)
         lines = header_section.split(b"\r\n")
-        m = re.search(r"\d{3}", lines[0].decode(errors="replace"))
+        m = RE_STATUS_CODE.search(lines[0].decode(errors="replace"))
         status = int(m.group()) if m else 0
         headers = {}
         for line in lines[1:]:

src/proxy_server.py

@@ -19,6 +19,10 @@
 
 log = logging.getLogger("Proxy")
 
+# Pre-compiled regexes for performance
+RE_MAX_AGE = re.compile(r"max-age=(\d+)")
+RE_CONTENT_TYPE = re.compile(r"content-type:\s*([^\r\n]+)")
+
 
 class ResponseCache:
     """Simple LRU response cache — avoids repeated relay calls."""
@@ -73,7 +77,7 @@ def parse_ttl(raw_response: bytes, url: str) -> int:
             return 0
 
         # Explicit max-age
-        m = re.search(r"max-age=(\d+)", hdr)
+        m = RE_MAX_AGE.search(hdr)
         if m:
             return min(int(m.group(1)), 86400)
 
@@ -88,7 +92,7 @@ def parse_ttl(raw_response: bytes, url: str) -> int:
             if path.endswith(ext):
                 return 3600  # 1 hour for static assets
 
-        ct_m = re.search(r"content-type:\s*([^\r\n]+)", hdr)
+        ct_m = RE_CONTENT_TYPE.search(hdr)
         ct = ct_m.group(1) if ct_m else ""
         if "image/" in ct or "font/" in ct:
             return 3600