didericis/bot-bottle
1
0
Fork 0
Code Issues 34 Pull Requests 11 Actions Packages Releases Wiki Activity

fix: bound heavy gateway operations

Browse Source
This commit is contained in:
didericis (codex)
2026-07-27 04:16:48 +00:00
committed by didericis
parent 78d3b43061
commit d879258f62
4 changed files with 95 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot_bottle/backend/cleanup_control.py
+13 -1
View File
@@ -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
+13 -3
View File
@@ -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
+32 -4
View File
@@ -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
+37 -24
View File
@@ -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: