From 4a13415b4398095d616ba0025e78d5e245be8ffa Mon Sep 17 00:00:00 2001 From: retoor Date: Thu, 3 Sep 2026 12:59:10 +0000 Subject: [PATCH] Fix blocking sync I/O on the event loop across containers, jobs, and xmlrpc The container/workspace reachability probes (socket connect + HTTP check) ran synchronously with real timeouts inside async request handlers and the live-view relay's 3-4s ticks, freezing the whole event loop whenever a container wasn't cleanly reachable - the likely cause of the periodic app-wide stalls. Converted the probe chain (api._port_reachable/_http_probe, editor_reachable, instance_runtime, provision.editor_ready/view) to real async I/O and parallelized the admin container/workspace list decorators. Also fixes: XmlrpcService.on_disable blocked up to 10s on a synchronous subprocess.wait inside an async method (now matches TelegramService's async-subprocess pattern); JobService._sweep_expired ran every job kind's cleanup() - including shutil.rmtree on large directories - synchronously on every tick, now offloaded via asyncio.to_thread for all job kinds at once; and several smaller blocking reads/writes on request/service paths (attachment-to-gitea mirroring, stealth chunked downloads, dbapi/isslop file reads, job payload/report I/O) moved off the loop thread. Co-Authored-By: Claude Sonnet 5 --- devplacepy/attachments.py | 2 +- devplacepy/routers/admin/containers.py | 2 +- devplacepy/routers/admin/workspaces.py | 11 +++-- devplacepy/routers/dbapi/query.py | 4 +- .../routers/projects/containers/instances.py | 2 +- .../routers/projects/containers/workspace.py | 4 +- devplacepy/routers/projects/index.py | 6 +-- devplacepy/routers/tools/isslop.py | 3 +- devplacepy/services/containers/api.py | 31 ++++++++----- .../containers/workspace/provision.py | 8 ++-- devplacepy/services/dbapi/service.py | 8 ++-- .../services/devii/container/controller.py | 2 +- .../services/devii/workspace/controller.py | 15 +++--- devplacepy/services/jobs/base.py | 6 +-- .../services/jobs/deepsearch/service.py | 26 ++++++----- devplacepy/services/jobs/isslop/service.py | 10 ++-- devplacepy/services/jobs/planning_service.py | 5 +- devplacepy/services/jobs/seo/service.py | 13 +++--- devplacepy/services/jobs/zip_service.py | 4 +- devplacepy/services/live_view_relay.py | 4 +- devplacepy/services/xmlrpc/__init__.py | 46 ++++++++++--------- devplacepy/stealth.py | 9 ++-- .../containers/workspace/provision.py | 15 +++--- 23 files changed, 132 insertions(+), 104 deletions(-) diff --git a/devplacepy/attachments.py b/devplacepy/attachments.py index c8016c8..d2862e8 100644 --- a/devplacepy/attachments.py +++ b/devplacepy/attachments.py @@ -501,7 +501,7 @@ async def mirror_attachment_to_gitea(uid): return None path = ATTACHMENTS_DIR / row.get("directory", "") / row.get("stored_name", "") try: - data = path.read_bytes() + data = await asyncio.to_thread(path.read_bytes) except OSError as exc: logger.warning("Cannot read attachment %s for Gitea mirror: %s", uid, exc) return None diff --git a/devplacepy/routers/admin/containers.py b/devplacepy/routers/admin/containers.py index 878267e..2a457aa 100644 --- a/devplacepy/routers/admin/containers.py +++ b/devplacepy/routers/admin/containers.py @@ -409,7 +409,7 @@ async def container_instance_page(request: Request, uid: str): "events": store.list_events(inst["uid"]), "schedules": store.list_schedules(inst["uid"]), "stats": api.instance_stats(inst["uid"]), - "runtime": api.instance_runtime(inst), + "runtime": await api.instance_runtime(inst), "can_manage": can_manage, "admin_section": "containers", }, diff --git a/devplacepy/routers/admin/workspaces.py b/devplacepy/routers/admin/workspaces.py index 39585cd..6aa7e0e 100644 --- a/devplacepy/routers/admin/workspaces.py +++ b/devplacepy/routers/admin/workspaces.py @@ -1,5 +1,6 @@ # retoor +import asyncio from typing import Annotated from fastapi import APIRouter, Depends, Request @@ -31,14 +32,14 @@ from devplacepy.utils import create_notification, generate_uid, not_found, requi router = APIRouter() -def _decorate(rows: list[dict]) -> list[dict]: +async def _decorate(rows: list[dict]) -> list[dict]: owner_uids = {row.get("workspace_owner_uid") for row in rows if row.get("workspace_owner_uid")} owners = get_users_by_uids(list(owner_uids)) if owner_uids else {} project_uids = {row.get("project_uid") for row in rows if row.get("project_uid")} projects = get_projects_by_uids(list(project_uids)) if project_uids else {} + views = await asyncio.gather(*(provision.view(row) for row in rows)) decorated = [] - for row in rows: - view = provision.view(row) + for row, view in zip(rows, views): owner = owners.get(row.get("workspace_owner_uid", "")) or {} project = projects.get(row.get("project_uid", "")) or {} view["owner_username"] = owner.get("username", "") @@ -79,7 +80,7 @@ async def admin_workspaces(request: Request): if not isinstance(admin, dict): return admin context = { - "workspaces": _decorate(_all_workspaces()), + "workspaces": await _decorate(_all_workspaces()), "flags": flags.list_flags(), "admin_section": "workspaces", "user": admin, @@ -104,7 +105,7 @@ async def admin_workspaces_data(request: Request): if not isinstance(admin, dict): return admin return JSONResponse( - {"workspaces": _decorate(_all_workspaces()), "flags": flags.list_flags()} + {"workspaces": await _decorate(_all_workspaces()), "flags": flags.list_flags()} ) diff --git a/devplacepy/routers/dbapi/query.py b/devplacepy/routers/dbapi/query.py index b5b715a..d3feb0b 100644 --- a/devplacepy/routers/dbapi/query.py +++ b/devplacepy/routers/dbapi/query.py @@ -1,5 +1,6 @@ # retoor +import asyncio import json import logging @@ -141,7 +142,8 @@ async def dbapi_query_result(request: Request, uid: str): if not path.is_relative_to(DBAPI_DIR.resolve()) or not path.is_file(): return error(404, "Result not available") queue.touch_job(uid, get_int_setting("dbquery_retention_seconds", 604800)) - return JSONResponse(json.loads(path.read_text(encoding="utf-8"))) + text = await asyncio.to_thread(path.read_text, encoding="utf-8") + return JSONResponse(json.loads(text)) @router.websocket("/query/{uid}/ws") diff --git a/devplacepy/routers/projects/containers/instances.py b/devplacepy/routers/projects/containers/instances.py index f925137..364af00 100644 --- a/devplacepy/routers/projects/containers/instances.py +++ b/devplacepy/routers/projects/containers/instances.py @@ -131,7 +131,7 @@ async def instance_detail(request: Request, project_slug: str, uid: str): "events": store.list_events(uid), "schedules": store.list_schedules(uid), "stats": api.instance_stats(uid), - "runtime": api.instance_runtime(inst), + "runtime": await api.instance_runtime(inst), } ) diff --git a/devplacepy/routers/projects/containers/workspace.py b/devplacepy/routers/projects/containers/workspace.py index 6b11791..41d2a05 100644 --- a/devplacepy/routers/projects/containers/workspace.py +++ b/devplacepy/routers/projects/containers/workspace.py @@ -74,7 +74,7 @@ async def workspace_page(request: Request, slug: str): profile = editor.resolve(user["uid"], instance) context = { "project": project, - "workspace": provision.view(instance) if instance else None, + "workspace": await provision.view(instance) if instance else None, "has_workspace": bool(instance), "viewer_can_workspace": True, "workspace_count": provision.count_for_owner(user["uid"]), @@ -130,7 +130,7 @@ async def workspace_open(request: Request, slug: str): ) provision.write_manifest(instance) return action_result( - request, f"/projects/{slug}/workspace", data=provision.view(instance) + request, f"/projects/{slug}/workspace", data=await provision.view(instance) ) diff --git a/devplacepy/routers/projects/index.py b/devplacepy/routers/projects/index.py index 6c5c7c0..f9707b7 100644 --- a/devplacepy/routers/projects/index.py +++ b/devplacepy/routers/projects/index.py @@ -187,12 +187,12 @@ async def projects_page( model=ProjectsOut, ) -def _editor_launch(project: dict, user: dict) -> dict: +async def _editor_launch(project: dict, user: dict) -> dict: from devplacepy.services.containers.workspace import editor, provision blank = {"url": "", "mode": "tab", "width": 0, "height": 0} instance = provision.find_for_project(project["uid"], user["uid"]) - if not instance or not provision.editor_ready(instance): + if not instance or not await provision.editor_ready(instance): return blank slug = project["slug"] or project["uid"] profile = editor.resolve(user["uid"], instance) @@ -262,7 +262,7 @@ async def project_detail(request: Request, project_slug: str, before: str = None ) viewer_can_workspace = can_open_workspace(project, user) editor_launch = ( - _editor_launch(project, user) + await _editor_launch(project, user) if viewer_can_workspace else {"url": "", "mode": "tab", "width": 0, "height": 0} ) diff --git a/devplacepy/routers/tools/isslop.py b/devplacepy/routers/tools/isslop.py index 68a44fe..b1eaf7f 100644 --- a/devplacepy/routers/tools/isslop.py +++ b/devplacepy/routers/tools/isslop.py @@ -1,5 +1,6 @@ # retoor +import asyncio import logging import re from urllib.parse import quote @@ -435,7 +436,7 @@ async def isslop_source(request: Request, uid: str, path: str, line: int = 0): source_path = (store.media_dir_for(uid) / source_name).resolve() if not source_path.is_relative_to(ISSLOP_MEDIA_DIR.resolve()) or not source_path.is_file(): raise not_found("Source not available for this file") - text = source_path.read_text(encoding="utf-8", errors="replace") + text = await asyncio.to_thread(source_path.read_text, encoding="utf-8", errors="replace") signals = store.decode_json(result.get("signals"), []) marked: dict[int, list] = {} for signal in signals: diff --git a/devplacepy/services/containers/api.py b/devplacepy/services/containers/api.py index d5e84ee..f964900 100644 --- a/devplacepy/services/containers/api.py +++ b/devplacepy/services/containers/api.py @@ -674,26 +674,33 @@ def instance_stats(instance_uid: str) -> dict: } -def _port_reachable(host: str, port: int, timeout: float = 0.3) -> bool: +async def _port_reachable(host: str, port: int, timeout: float = 0.3) -> bool: try: - with socket.create_connection((host, port), timeout=timeout): - return True - except OSError: + _, writer = await asyncio.wait_for( + asyncio.open_connection(host, port), timeout=timeout + ) + except (OSError, asyncio.TimeoutError): return False + writer.close() + try: + await writer.wait_closed() + except OSError: + pass + return True -def editor_reachable(instance: dict) -> bool: +async def editor_reachable(instance: dict) -> bool: port = int(instance.get("editor_port") or 0) host, target_port = tunnel_target(instance, port) if not host or not target_port: return False - return _port_reachable(host, target_port) + return await _port_reachable(host, target_port) -def _http_probe(host: str, port: int, timeout: float = 1.0) -> str: +async def _http_probe(host: str, port: int, timeout: float = 1.0) -> str: try: - with stealth.stealth_sync_client(timeout=timeout) as client: - response = client.get(f"http://{host}:{port}/") + async with stealth.stealth_async_client(timeout=timeout) as client: + response = await client.get(f"http://{host}:{port}/") return f"HTTP {response.status_code}" except Exception as exc: # noqa: BLE001 - diagnostic, any failure is informative return f"unreachable: {type(exc).__name__}" @@ -760,7 +767,7 @@ def tunnel_target(instance: dict, container_port: int) -> tuple: return reachable_target(instance, container_port, port_maps) -def instance_runtime(instance: dict) -> dict: +async def instance_runtime(instance: dict) -> dict: boot = (instance.get("boot_command") or "").strip() port_maps = json.loads(instance.get("ports_json") or "[]") container_ip = (instance.get("container_ip") or "").strip() @@ -775,13 +782,13 @@ def instance_runtime(instance: dict) -> dict: "container": int(mapping.get("container") or 0), "host": host_port, "proto": mapping.get("proto", "tcp"), - "reachable": _port_reachable(probe_host, host_port) + "reachable": await _port_reachable(probe_host, host_port) if host_port else False, } ) ingress_serving = ( - _http_probe(target_host, target_port) + await _http_probe(target_host, target_port) if target_host and target_port else "no ingress port mapped" ) diff --git a/devplacepy/services/containers/workspace/provision.py b/devplacepy/services/containers/workspace/provision.py index a5e8d7f..c68d8c6 100644 --- a/devplacepy/services/containers/workspace/provision.py +++ b/devplacepy/services/containers/workspace/provision.py @@ -241,12 +241,12 @@ def write_manifest(instance: dict) -> None: return -def editor_ready(instance: dict) -> bool: +async def editor_ready(instance: dict) -> bool: if instance.get("suspended_at"): return False if instance.get("status") != store.ST_RUNNING: return False - return api.editor_reachable(instance) + return await api.editor_reachable(instance) def phase(instance: dict, ready: bool) -> str: @@ -264,12 +264,12 @@ def phase(instance: dict, ready: bool) -> str: return PHASE_STOPPED -def view(instance: dict) -> dict: +async def view(instance: dict) -> dict: owner_uid = instance.get("workspace_owner_uid", "") limits = quota.resolve(owner_uid, instance) disk_used = int(instance.get("disk_bytes") or 0) egress_used = int(instance.get("egress_bytes") or 0) - ready = editor_ready(instance) + ready = await editor_ready(instance) current = phase(instance, ready) return { "uid": instance.get("uid", ""), diff --git a/devplacepy/services/dbapi/service.py b/devplacepy/services/dbapi/service.py index e749398..a36ced4 100644 --- a/devplacepy/services/dbapi/service.py +++ b/devplacepy/services/dbapi/service.py @@ -88,7 +88,7 @@ class DbApiJobService(JobService): rows, truncated = await self._stream(uid, verdict.sql, max_rows) output_dir = self.result_dir(uid) - output_dir.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(output_dir.mkdir, parents=True, exist_ok=True) result = { "sql": verdict.sql, "row_count": len(rows), @@ -96,8 +96,10 @@ class DbApiJobService(JobService): "suspicious": verdict.suspicious, "rows": rows, } - (output_dir / "result.json").write_text( - json.dumps(result, default=str), encoding="utf-8" + await asyncio.to_thread( + (output_dir / "result.json").write_text, + json.dumps(result, default=str), + encoding="utf-8", ) hub.publish( uid, diff --git a/devplacepy/services/devii/container/controller.py b/devplacepy/services/devii/container/controller.py index f9d4248..a2eee64 100644 --- a/devplacepy/services/devii/container/controller.py +++ b/devplacepy/services/devii/container/controller.py @@ -225,7 +225,7 @@ class ContainerController: "instance": inst["name"], "status": inst["status"], "ingress_url": self._ingress_url(inst), - "runtime": api.instance_runtime(inst), + "runtime": await api.instance_runtime(inst), "stats": api.instance_stats(inst["uid"]), }, default=str, diff --git a/devplacepy/services/devii/workspace/controller.py b/devplacepy/services/devii/workspace/controller.py index c944465..6de32da 100644 --- a/devplacepy/services/devii/workspace/controller.py +++ b/devplacepy/services/devii/workspace/controller.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import json from typing import Any @@ -73,23 +74,23 @@ class WorkspaceController: instance = await provision.ensure(project, user) instance = provision.resume(instance) provision.write_manifest(instance) - view = provision.view(instance) + view = await provision.view(instance) view["editor_url"] = ( f"/projects/{project.get('slug') or project['uid']}/workspace" ) return view - def _workspace_status(self, args: dict) -> Any: - return provision.view(self._resolve(args.get("project_slug", ""))) + async def _workspace_status(self, args: dict) -> Any: + return await provision.view(self._resolve(args.get("project_slug", ""))) - def _workspace_list(self, args: dict) -> Any: + async def _workspace_list(self, args: dict) -> Any: table = get_table("instances") filters: dict[str, Any] = {"is_workspace": 1, "deleted_at": None} if not self.admin: filters["workspace_owner_uid"] = self.owner_id - return { - "workspaces": [provision.view(row) for row in table.find(**filters)] - } + rows = list(table.find(**filters)) + views = await asyncio.gather(*(provision.view(row) for row in rows)) + return {"workspaces": list(views)} def _workspace_stop(self, args: dict) -> Any: instance = self._resolve(args.get("project_slug", "")) diff --git a/devplacepy/services/jobs/base.py b/devplacepy/services/jobs/base.py index fcfaa5f..600cdeb 100644 --- a/devplacepy/services/jobs/base.py +++ b/devplacepy/services/jobs/base.py @@ -86,7 +86,7 @@ class JobService(BaseService): self._reap() self._recover_orphans(min_age_seconds=self.job_timeout_seconds()) self._refill() - self._sweep_expired() + await self._sweep_expired() def _reap(self) -> None: for uid in list(self._inflight): @@ -208,7 +208,7 @@ class JobService(BaseService): ) self.log(f"Recovered orphaned job {uid} (retry {retry_count + 1})") - def _sweep_expired(self) -> None: + async def _sweep_expired(self) -> None: table = get_table("jobs") now = datetime.now(timezone.utc) for row in table.find(kind=self.kind, status=queue.DONE): @@ -217,7 +217,7 @@ class JobService(BaseService): continue job = queue.get_job(row["uid"]) try: - self.cleanup(job) + await asyncio.to_thread(self.cleanup, job) except Exception as exc: self.log(f"Cleanup failed for {row['uid']}: {exc}") table.delete(uid=row["uid"]) diff --git a/devplacepy/services/jobs/deepsearch/service.py b/devplacepy/services/jobs/deepsearch/service.py index 19a1042..cfea2e6 100644 --- a/devplacepy/services/jobs/deepsearch/service.py +++ b/devplacepy/services/jobs/deepsearch/service.py @@ -59,15 +59,17 @@ class DeepsearchService(JobService): payload = dict(job.get("payload", {})) query = payload.get("query", "") output_dir = self.session_dir(uid) - output_dir.mkdir(parents=True, exist_ok=True) - (output_dir / "control.json").write_text( - json.dumps({"state": "running"}), encoding="utf-8" + await asyncio.to_thread(output_dir.mkdir, parents=True, exist_ok=True) + await asyncio.to_thread( + (output_dir / "control.json").write_text, + json.dumps({"state": "running"}), + encoding="utf-8", ) payload["collection"] = self.collection_name(uid) payload["model"] = self.get_config()["deepsearch_model"] or INTERNAL_MODEL payload["cached_hashes"] = self._cached_hashes(database) payload_path = output_dir / "payload.json" - payload_path.write_text(json.dumps(payload), encoding="utf-8") + await asyncio.to_thread(payload_path.write_text, json.dumps(payload), encoding="utf-8") actor_kind = job.get("owner_kind") or "system" actor_uid = job.get("owner_id") if job.get("owner_kind") == "user" else None @@ -89,11 +91,11 @@ class DeepsearchService(JobService): metadata={"query": query, "error": str(exc)[:200]}, links=[audit.job(uid)], ) - shutil.rmtree(output_dir, ignore_errors=True) + await asyncio.to_thread(shutil.rmtree, output_dir, ignore_errors=True) raise - report = self._load_report(output_dir) - self._persist_cache(database, output_dir) + report = await self._load_report(output_dir) + await self._persist_cache(database, output_dir) from datetime import datetime, timezone database.update_deepsearch_session( @@ -158,12 +160,13 @@ class DeepsearchService(JobService): if row.get("url_hash") ] - def _persist_cache(self, database, output_dir: Path) -> None: + async def _persist_cache(self, database, output_dir: Path) -> None: path = output_dir / "url_cache.json" if not path.is_file(): return try: - entries = json.loads(path.read_text(encoding="utf-8")) + text = await asyncio.to_thread(path.read_text, encoding="utf-8") + entries = json.loads(text) except (ValueError, OSError): return for entry in entries: @@ -244,12 +247,13 @@ class DeepsearchService(JobService): ) return summary - def _load_report(self, output_dir: Path) -> dict: + async def _load_report(self, output_dir: Path) -> dict: report_path = output_dir / "report.json" if not report_path.is_file(): return {} try: - return json.loads(report_path.read_text(encoding="utf-8")) + text = await asyncio.to_thread(report_path.read_text, encoding="utf-8") + return json.loads(text) except (ValueError, OSError): return {} diff --git a/devplacepy/services/jobs/isslop/service.py b/devplacepy/services/jobs/isslop/service.py index 1e1cbf3..68b062e 100644 --- a/devplacepy/services/jobs/isslop/service.py +++ b/devplacepy/services/jobs/isslop/service.py @@ -141,21 +141,23 @@ class IsslopService(JobService): ) run_dir = self.run_dir(uid) - run_dir.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(run_dir.mkdir, parents=True, exist_ok=True) payload_path = run_dir / "payload.json" - payload_path.write_text(json.dumps(self._worker_payload(job)), encoding="utf-8") + await asyncio.to_thread( + payload_path.write_text, json.dumps(self._worker_payload(job)), encoding="utf-8" + ) try: workspace = workspace_for(ISSLOP_WORKSPACES_DIR, source_url, uid) except ValueError as error: await self._relay(persister, WorkerEvent(KIND_ERROR, f"Workspace rejected: {error}", {})) - shutil.rmtree(run_dir, ignore_errors=True) + await asyncio.to_thread(shutil.rmtree, run_dir, ignore_errors=True) raise RuntimeError(str(error)) try: final = await self._run_worker(uid, persister, payload_path, workspace) finally: await asyncio.to_thread(remove_workspace, workspace) - shutil.rmtree(run_dir, ignore_errors=True) + await asyncio.to_thread(shutil.rmtree, run_dir, ignore_errors=True) if final.kind == KIND_ERROR: audit.record_system( diff --git a/devplacepy/services/jobs/planning_service.py b/devplacepy/services/jobs/planning_service.py index 782dd09..ddeb1a7 100644 --- a/devplacepy/services/jobs/planning_service.py +++ b/devplacepy/services/jobs/planning_service.py @@ -1,5 +1,6 @@ # retoor +import asyncio import logging import zlib from pathlib import Path @@ -52,9 +53,9 @@ class PlanningReportService(JobService): final_name = self._final_name(job.get("preferred_name", ""), markdown) target_dir = PLANNING_REPORTS_DIR / _directory_for(uid) - target_dir.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(target_dir.mkdir, parents=True, exist_ok=True) final_path = target_dir / final_name - final_path.write_text(markdown, encoding="utf-8") + await asyncio.to_thread(final_path.write_text, markdown, encoding="utf-8") bytes_out = len(markdown.encode("utf-8")) except Exception: audit.record_system( diff --git a/devplacepy/services/jobs/seo/service.py b/devplacepy/services/jobs/seo/service.py index 970d38a..75ae117 100644 --- a/devplacepy/services/jobs/seo/service.py +++ b/devplacepy/services/jobs/seo/service.py @@ -41,9 +41,9 @@ class SeoService(JobService): payload = job.get("payload", {}) target = payload.get("url", "") output_dir = self.report_dir(uid) - output_dir.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(output_dir.mkdir, parents=True, exist_ok=True) payload_path = output_dir / "payload.json" - payload_path.write_text(json.dumps(payload), encoding="utf-8") + await asyncio.to_thread(payload_path.write_text, json.dumps(payload), encoding="utf-8") actor_kind = "user" if job.get("owner_kind") == "user" else ( job.get("owner_kind") or "system" @@ -63,10 +63,10 @@ class SeoService(JobService): metadata={"target": target, "error": str(exc)[:200]}, links=[audit.job(uid)], ) - shutil.rmtree(output_dir, ignore_errors=True) + await asyncio.to_thread(shutil.rmtree, output_dir, ignore_errors=True) raise - report = self._load_report(output_dir) + report = await self._load_report(output_dir) hub.publish( uid, { @@ -154,12 +154,13 @@ class SeoService(JobService): ) return summary - def _load_report(self, output_dir: Path) -> dict: + async def _load_report(self, output_dir: Path) -> dict: report_path = output_dir / "report.json" if not report_path.is_file(): return {} try: - return json.loads(report_path.read_text(encoding="utf-8")) + text = await asyncio.to_thread(report_path.read_text, encoding="utf-8") + return json.loads(text) except (ValueError, OSError): return {} diff --git a/devplacepy/services/jobs/zip_service.py b/devplacepy/services/jobs/zip_service.py index 241f39c..caef254 100644 --- a/devplacepy/services/jobs/zip_service.py +++ b/devplacepy/services/jobs/zip_service.py @@ -41,12 +41,12 @@ class ZipService(JobService): try: item_count = await asyncio.to_thread(self._materialize, source, staging) tmp_dir = ZIPS_DIR / _directory_for(uid) - tmp_dir.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(tmp_dir.mkdir, parents=True, exist_ok=True) tmp_zip = tmp_dir / f"{generate_uid()}.zip" stats = await self._run_worker(staging, tmp_zip) final_name = self._final_name(job.get("preferred_name", ""), stats["crc32"]) final_path = tmp_dir / final_name - tmp_zip.replace(final_path) + await asyncio.to_thread(tmp_zip.replace, final_path) except Exception: audit.record_system( "job.zip.failed", diff --git a/devplacepy/services/live_view_relay.py b/devplacepy/services/live_view_relay.py index 33ce75f..d1e302d 100644 --- a/devplacepy/services/live_view_relay.py +++ b/devplacepy/services/live_view_relay.py @@ -58,7 +58,7 @@ async def _container_detail(match: re.Match) -> Optional[dict]: "events": store.list_events(uid), "schedules": store.list_schedules(uid), "stats": api.instance_stats(uid), - "runtime": api.instance_runtime(inst), + "runtime": await api.instance_runtime(inst), } @@ -128,7 +128,7 @@ async def _workspace_detail(match: re.Match) -> Optional[dict]: project = _instance_project(inst) slug = project.get("slug") or project.get("uid") or "" return { - "workspace": provision.view(inst), + "workspace": await provision.view(inst), "editor_url": ( f"/projects/{slug}/containers/instances/{inst['uid']}/code/" if slug else "" ), diff --git a/devplacepy/services/xmlrpc/__init__.py b/devplacepy/services/xmlrpc/__init__.py index 6808853..cfe1017 100644 --- a/devplacepy/services/xmlrpc/__init__.py +++ b/devplacepy/services/xmlrpc/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations -import subprocess +import asyncio import sys from devplacepy.config import XMLRPC_BIND, XMLRPC_PORT @@ -24,48 +24,50 @@ class XmlrpcService(BaseService): def __init__(self) -> None: super().__init__("xmlrpc", interval_seconds=XMLRPC_INTERVAL_SECONDS) - self._process: subprocess.Popen | None = None + self._process: asyncio.subprocess.Process | None = None def _alive(self) -> bool: - return self._process is not None and self._process.poll() is None + return self._process is not None and self._process.returncode is None - def _spawn(self) -> None: - self._process = subprocess.Popen( - [sys.executable, "-m", SERVER_MODULE], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, + async def _spawn(self) -> None: + self._process = await asyncio.create_subprocess_exec( + sys.executable, + "-m", + SERVER_MODULE, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, ) self.log( f"Forking XML-RPC server started (pid {self._process.pid}) on " f"{XMLRPC_BIND}:{XMLRPC_PORT}" ) - def _terminate(self) -> None: - if not self._alive(): - self._process = None - return - self._process.terminate() - try: - self._process.wait(timeout=TERMINATE_TIMEOUT_SECONDS) - except subprocess.TimeoutExpired: - self._process.kill() - self._process.wait() - self.log("Forking XML-RPC server stopped") + async def _terminate(self) -> None: + process = self._process self._process = None + if process is None or process.returncode is not None: + return + process.terminate() + try: + await asyncio.wait_for(process.wait(), timeout=TERMINATE_TIMEOUT_SECONDS) + except asyncio.TimeoutError: + process.kill() + await process.wait() + self.log("Forking XML-RPC server stopped") async def on_enable(self) -> None: if not self._alive(): - self._spawn() + await self._spawn() async def on_disable(self) -> None: - self._terminate() + await self._terminate() async def run_once(self) -> None: if self._alive(): self.log(f"XML-RPC server healthy (pid {self._process.pid})") return self.log("XML-RPC server not running, starting it") - self._spawn() + await self._spawn() def collect_metrics(self) -> dict: return { diff --git a/devplacepy/stealth.py b/devplacepy/stealth.py index 8e96e0b..e82db4d 100644 --- a/devplacepy/stealth.py +++ b/devplacepy/stealth.py @@ -494,7 +494,7 @@ class ChromeStealthClient: dest_type: str = "empty", ) -> DownloadResult: directory = Path(destination) - directory.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(directory.mkdir, parents=True, exist_ok=True) try: target = resolve_destination(directory, url, filename) except ValueError as error: @@ -505,10 +505,13 @@ class ChromeStealthClient: try: async with self.stream("GET", url, referer=referer, dest=dest_type) as response: response.raise_for_status() - with target.open("wb") as handle: + handle = await asyncio.to_thread(target.open, "wb") + try: async for chunk in response.aiter_bytes(chunk_size): - handle.write(chunk) + await asyncio.to_thread(handle.write, chunk) written += len(chunk) + finally: + await asyncio.to_thread(handle.close) logger.info("Downloaded %s -> %s (%d bytes)", url, target, written) return DownloadResult( url=url, diff --git a/tests/unit/services/containers/workspace/provision.py b/tests/unit/services/containers/workspace/provision.py index f0f1521..45ad8c4 100644 --- a/tests/unit/services/containers/workspace/provision.py +++ b/tests/unit/services/containers/workspace/provision.py @@ -1,5 +1,6 @@ # retoor +import asyncio import itertools import socket @@ -117,7 +118,7 @@ def test_transitional_phases_are_exactly_starting_and_stopping(): def test_editor_ready_when_the_editor_port_accepts_connections(listener): port = listener.getsockname()[1] row = _instance(container_ip="127.0.0.1", editor_port=port) - assert provision.editor_ready(row) is True + assert asyncio.run(provision.editor_ready(row)) is True def test_editor_ready_is_false_when_nothing_listens(): @@ -126,12 +127,12 @@ def test_editor_ready_is_false_when_nothing_listens(): port = probe.getsockname()[1] probe.close() row = _instance(container_ip="127.0.0.1", editor_port=port) - assert provision.editor_ready(row) is False + assert asyncio.run(provision.editor_ready(row)) is False def test_editor_ready_is_false_without_a_reachable_target(listener): row = _instance(container_ip="", ports_json="[]") - assert provision.editor_ready(row) is False + assert asyncio.run(provision.editor_ready(row)) is False def test_editor_ready_is_false_unless_the_container_runs(listener): @@ -140,7 +141,7 @@ def test_editor_ready_is_false_unless_the_container_runs(listener): if status == store.ST_RUNNING: continue row = _instance(container_ip="127.0.0.1", editor_port=port, status=status) - assert provision.editor_ready(row) is False + assert asyncio.run(provision.editor_ready(row)) is False def test_editor_ready_is_false_while_suspended(listener): @@ -148,7 +149,7 @@ def test_editor_ready_is_false_while_suspended(listener): row = _instance( container_ip="127.0.0.1", editor_port=port, suspended_at="2026-01-01T00:00:00" ) - assert provision.editor_ready(row) is False + assert asyncio.run(provision.editor_ready(row)) is False def test_view_carries_the_phase_its_label_readiness_and_the_owner(listener): @@ -166,13 +167,13 @@ def test_view_carries_the_phase_its_label_readiness_and_the_owner(listener): "ports_json": "[]", } ) - view = provision.view(instance) + view = asyncio.run(provision.view(instance)) assert view["phase"] == provision.PHASE_READY assert view["phase_label"] == "Ready" assert view["editor_ready"] is True assert view["owner_uid"] == OWNER store.update_instance(instance["uid"], {"desired_state": store.DESIRED_STOPPED}) - view = provision.view(store.get_instance(instance["uid"])) + view = asyncio.run(provision.view(store.get_instance(instance["uid"]))) assert view["phase"] == provision.PHASE_STOPPING assert view["editor_ready"] is True