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 <noreply@anthropic.com>

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

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",
         },

devplacepy/routers/admin/workspaces.py

@@ -1,5 +1,6 @@
 # retoor <retoor@molodetz.nl>
 
+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()}
     )
 
 

devplacepy/routers/dbapi/query.py

@@ -1,5 +1,6 @@
 # retoor <retoor@molodetz.nl>
 
+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")

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),
         }
     )
 

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)
     )
 
 

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}
     )

devplacepy/routers/tools/isslop.py

@@ -1,5 +1,6 @@
 # retoor <retoor@molodetz.nl>
 
+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:

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"
     )

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", ""),

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,

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,

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", ""))

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"])

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 {}
 

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(

devplacepy/services/jobs/planning_service.py

@@ -1,5 +1,6 @@
 # retoor <retoor@molodetz.nl>
 
+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(

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 {}
 

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",

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 ""
         ),

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():
+    async def _terminate(self) -> None:
+        process = self._process
         self._process = None
+        if process is None or process.returncode is not None:
             return
-        self._process.terminate()
+        process.terminate()
         try:
-            self._process.wait(timeout=TERMINATE_TIMEOUT_SECONDS)
-        except subprocess.TimeoutExpired:
-            self._process.kill()
-            self._process.wait()
+            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")
-        self._process = None
 
     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 {

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,

tests/unit/services/containers/workspace/provision.py

@@ -1,5 +1,6 @@
 # retoor <retoor@molodetz.nl>
 
+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