From d879258f6257bff1a13243029e026721c76bfc4a Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 27 Jul 2026 04:16:48 +0000 Subject: [PATCH] fix: bound heavy gateway operations --- bot_bottle/backend/cleanup_control.py | 14 ++++- bot_bottle/backend/firecracker/cleanup.py | 16 +++++- bot_bottle/gateway/bounded_http.py | 36 ++++++++++-- bot_bottle/gateway/git_gate/http_backend.py | 61 +++++++++++++-------- 4 files changed, 95 insertions(+), 32 deletions(-) diff --git a/bot_bottle/backend/cleanup_control.py b/bot_bottle/backend/cleanup_control.py index b82efd6f..639df4a8 100644 --- a/bot_bottle/backend/cleanup_control.py +++ b/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)) diff --git a/bot_bottle/backend/firecracker/cleanup.py b/bot_bottle/backend/firecracker/cleanup.py index 0e560684..5e4221e8 100644 --- a/bot_bottle/backend/firecracker/cleanup.py +++ b/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) diff --git a/bot_bottle/gateway/bounded_http.py b/bot_bottle/gateway/bounded_http.py index 730b70b4..08be3c99 100644 --- a/bot_bottle/gateway/bounded_http.py +++ b/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", +] diff --git a/bot_bottle/gateway/git_gate/http_backend.py b/bot_bottle/gateway/git_gate/http_backend.py index 4ac3eb39..d9538f5d 100644 --- a/bot_bottle/gateway/git_gate/http_backend.py +++ b/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: