fix: bound heavy gateway operations

bot_bottle/backend/cleanup_control.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import shutil
 import subprocess
 from collections.abc import Sequence
@@ -19,11 +20,19 @@ class CleanupFailures:
         self._messages: list[str] = []
 
     def run(self, argv: Sequence[str], description: str) -> None:
+        raw_timeout = os.environ.get(
+            "BOT_BOTTLE_CLEANUP_COMMAND_TIMEOUT_SECONDS", "120",
+        )
+        try:
+            timeout = float(raw_timeout)
+        except ValueError:
+            timeout = 120.0
         try:
             result = subprocess.run(
                 list(argv), capture_output=True, text=True, check=False,
+                timeout=max(timeout, 1.0),
             )
-        except OSError as exc:
+        except (OSError, subprocess.SubprocessError) as exc:
             self._messages.append(f"{description}: {exc}")
             return
         if result.returncode != 0:
@@ -40,6 +49,9 @@ class CleanupFailures:
         except OSError as exc:
             self._messages.append(f"{description}: {exc}")
 
+    def record(self, message: str) -> None:
+        self._messages.append(message)
+
     def raise_if_any(self) -> None:
         if self._messages:
             raise CleanupError("; ".join(self._messages))

bot_bottle/backend/firecracker/cleanup.py

@@ -29,7 +29,7 @@ from pathlib import Path
 
 from ...log import info
 from .. import EnumerationError
-from ..cleanup_control import CleanupFailures
+from ..cleanup_control import CleanupError, CleanupFailures
 from . import lifecycle_lock, util
 from .bottle_cleanup_plan import FirecrackerBottleCleanupPlan
 
@@ -153,7 +153,10 @@ def cleanup(plan: FirecrackerBottleCleanupPlan) -> None:
         approved_dirs = set(plan.run_dirs).intersection(fresh.run_dirs)
         failures = CleanupFailures()
         for pid in sorted(approved_pids):
-            _terminate_orphan(pid, _run_root())
+            try:
+                _terminate_orphan(pid, _run_root())
+            except CleanupError as exc:
+                failures.record(str(exc))
         for path in sorted(approved_dirs):
             info(f"rm -rf {path}")
             failures.remove_tree(Path(path), f"removing Firecracker run dir {path}")
@@ -184,6 +187,13 @@ def _terminate_orphan(pid: int, run_root: Path) -> None:
         if run_dir is None or run_dir.is_dir():
             return
         info(f"kill firecracker VM pid {pid}")
-        signal.pidfd_send_signal(pidfd, signal.SIGTERM)
+        try:
+            signal.pidfd_send_signal(pidfd, signal.SIGTERM)
+        except ProcessLookupError:
+            return
+        except OSError as exc:
+            raise CleanupError(
+                f"could not signal Firecracker pid {pid}: {exc}"
+            ) from exc
     finally:
         os.close(pidfd)

bot_bottle/gateway/bounded_http.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import http.server
+import io
 import socket
 import threading
 import time
@@ -14,6 +15,10 @@ class Readable(Protocol):
     def read(self, size: int = -1, /) -> bytes: ...
 
 
+class Writable(Protocol):
+    def write(self, data: bytes, /) -> object: ...
+
+
 @dataclass(frozen=True)
 class BodyReadError(Exception):
     status: int
@@ -30,6 +35,25 @@ def read_declared_body(
     require_length: bool,
 ) -> bytes:
     """Validate and read exactly one declared body under a read deadline."""
+    output = io.BytesIO()
+    copy_declared_body(
+        stream, output, connection, raw_length, maximum=maximum,
+        timeout_seconds=timeout_seconds, require_length=require_length,
+    )
+    return output.getvalue()
+
+
+def copy_declared_body(
+    stream: Readable,
+    output: Writable,
+    connection: socket.socket,
+    raw_length: str | None,
+    *,
+    maximum: int,
+    timeout_seconds: float,
+    require_length: bool,
+) -> int:
+    """Copy one declared body to a sink without retaining it in memory."""
     if raw_length is None:
         if require_length:
             raise BodyReadError(411, "Content-Length required")
@@ -44,7 +68,6 @@ def read_declared_body(
         raise BodyReadError(413, "request body too large")
     previous_timeout = connection.gettimeout()
     deadline = time.monotonic() + timeout_seconds
-    chunks: list[bytes] = []
     remaining = length
     try:
         while remaining:
@@ -55,13 +78,13 @@ def read_declared_body(
             chunk = stream.read(min(remaining, 64 * 1024))
             if not chunk:
                 raise BodyReadError(400, "incomplete request body")
-            chunks.append(chunk)
+            output.write(chunk)
             remaining -= len(chunk)
     except TimeoutError as exc:
         raise BodyReadError(408, "request body read timed out") from exc
     finally:
         connection.settimeout(previous_timeout)
-    return b"".join(chunks)
+    return length
 
 
 class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer):
@@ -106,4 +129,9 @@ class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer):
             self._request_slots.release()
 
 
-__all__ = ["BodyReadError", "BoundedThreadingHTTPServer", "read_declared_body"]
+__all__ = [
+    "BodyReadError",
+    "BoundedThreadingHTTPServer",
+    "copy_declared_body",
+    "read_declared_body",
+]

bot_bottle/gateway/git_gate/http_backend.py

@@ -21,6 +21,8 @@ from __future__ import annotations
 import os
 import subprocess
 import sys
+import tempfile
+import threading
 import typing
 from http.server import BaseHTTPRequestHandler
 from pathlib import Path
@@ -30,7 +32,7 @@ from bot_bottle.constants import GIT_GATE_TIMEOUT_SECS, IDENTITY_HEADER
 from bot_bottle.gateway.bounded_http import (
     BodyReadError,
     BoundedThreadingHTTPServer,
-    read_declared_body,
+    copy_declared_body,
 )
 from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
 
@@ -84,6 +86,8 @@ def resolve_sandbox_root(
 MAX_BODY_BYTES = 100 * 1024 * 1024
 REQUEST_BODY_TIMEOUT_SECONDS = 30.0
 MAX_REQUEST_WORKERS = 16
+MAX_BODY_WORKERS = 2
+_BODY_WORK_SLOTS = threading.BoundedSemaphore(MAX_BODY_WORKERS)
 
 
 class GitHttpHandler(BaseHTTPRequestHandler):
@@ -191,31 +195,40 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             value = self.headers.get(header)
             if value:
                 env[variable] = value
-        try:
-            body = read_declared_body(
-                self.rfile,
-                self.connection,
-                self.headers.get("content-length"),
-                maximum=MAX_BODY_BYTES,
-                timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
-                require_length=False,
-            )
-        except BodyReadError as exc:
-            self.send_error(exc.status, exc.message)
+        if not _BODY_WORK_SLOTS.acquire(blocking=False):
+            self.send_error(503, "git request capacity exhausted")
             return
         try:
-            proc = subprocess.run(
-                ["git", "http-backend"],
-                input=body,
-                env=env,
-                capture_output=True,
-                check=False,
-                timeout=GIT_GATE_TIMEOUT_SECS,
-            )
-        except (OSError, subprocess.SubprocessError) as exc:
-            self.log_message("git http-backend unavailable: %s", exc)
-            self.send_error(503, "git backend unavailable")
-            return
+            with tempfile.TemporaryFile() as body:
+                try:
+                    copy_declared_body(
+                        self.rfile,
+                        body,
+                        self.connection,
+                        self.headers.get("content-length"),
+                        maximum=MAX_BODY_BYTES,
+                        timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
+                        require_length=False,
+                    )
+                except BodyReadError as exc:
+                    self.send_error(exc.status, exc.message)
+                    return
+                body.seek(0)
+                try:
+                    proc = subprocess.run(
+                        ["git", "http-backend"],
+                        stdin=body,
+                        env=env,
+                        capture_output=True,
+                        check=False,
+                        timeout=GIT_GATE_TIMEOUT_SECS,
+                    )
+                except (OSError, subprocess.SubprocessError) as exc:
+                    self.log_message("git http-backend unavailable: %s", exc)
+                    self.send_error(503, "git backend unavailable")
+                    return
+        finally:
+            _BODY_WORK_SLOTS.release()
         self._write_cgi_response(proc.stdout)
 
     def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None: