diff --git a/backend/src/backend/schedule_runs.py b/backend/src/backend/schedule_runs.py index ea3357d..6fbb131 100644 --- a/backend/src/backend/schedule_runs.py +++ b/backend/src/backend/schedule_runs.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio from datetime import UTC, datetime from typing import Any, Literal from urllib.parse import quote @@ -19,6 +18,7 @@ from pydantic import Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from common.config import settings from common.db.models import ( ScheduleNodeRuns, ScheduleRuns, @@ -199,7 +199,7 @@ async def _visible_artifact( ) if ( item is None - or item.storage_backend != "rustfs" + or item.storage_backend != settings.storage_backend or not item.bucket_name or not item.object_key ): @@ -223,10 +223,8 @@ async def _artifact_bytes( item: StorageObjects, request: Request, ) -> bytes: - return await asyncio.to_thread( - request.app.state.object_store.get_bytes, - bucket_name=item.bucket_name, - object_key=item.object_key, + return await request.app.state.object_stores[item.bucket_name].get( + item.object_key, ) diff --git a/docker-compose.yml b/docker-compose.yml index 4d5e487..4693ddb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -48,7 +48,6 @@ services: volumes: - ${PWD}:/app - ./default.conf:/etc/nginx/conf.d/default.conf.template:ro - - ./scripts/nginx-entrypoint.sh:/usr/local/bin/model-platform-entrypoint.sh:ro healthcheck: test: ["CMD-SHELL", "wget -qO- http://127.0.0.1/ >/dev/null"] interval: 10s @@ -66,6 +65,13 @@ services: max-size: "200m" max-file: "10" restart: unless-stopped + # The source tree is bind-mounted for local development while /app/.venv + # is an anonymous Docker volume. Sync first so restarts never keep an + # older installed backend package. + command: + - sh + - -c + - uv sync --no-dev --no-editable --package backend && uv pip install --python /app/.venv/bin/python --reinstall --no-deps ./common ./backend && exec gunicorn --config backend/gunicorn.conf.py backend.main:app # No host port: architecture §2.2 — only Nginx is externally reachable. # No local-FS volume: backend stores everything in S3 (S3_*). environment: @@ -95,7 +101,7 @@ services: runtime: condition: service_healthy ports: - - 8891:8000 + - 9121:8000 volumes: - ${PWD}:/app - ./data:/data @@ -118,7 +124,7 @@ services: max-file: "10" restart: unless-stopped ports: - - 8892:8000 + - 9122:8000 cap_add: - SYS_ADMIN devices: @@ -192,6 +198,8 @@ services: S3_WORKSPACE_BUCKET: ${S3_WORKSPACE_BUCKET:-workspace} S3_VERSION_BUCKET: ${S3_VERSION_BUCKET:-version} S3_RUN_LOG_BUCKET: ${S3_RUN_LOG_BUCKET:-run-log} + # Backend's health endpoint already verifies MySQL; the scheduler only + # needs MySQL and Backend to be ready in either local or S3 mode. READINESS_TARGETS: ${MYSQL_HOST:?MYSQL_HOST is required}:${MYSQL_PORT:-3306},backend:8000 depends_on: backend: diff --git a/runtime/src/runtime/main.py b/runtime/src/runtime/main.py index 9ff443e..e57a37d 100644 --- a/runtime/src/runtime/main.py +++ b/runtime/src/runtime/main.py @@ -7,6 +7,7 @@ surface is two endpoints; everything else is lifespan orchestration. from __future__ import annotations +import asyncio from contextlib import asynccontextmanager from typing import Literal @@ -48,7 +49,10 @@ async def lifespan(app: FastAPI): f"removed_meta={counters['removed_meta']} " f"live_in_registry={counters['live_in_registry']}" ) - await scan_workspaces() + # Scanning a remote FUSE mount may take longer than the HTTP health-check + # timeout. Keep recovery asynchronous so dependent services can start as + # soon as the runtime API is ready. + asyncio.create_task(scan_workspaces(), name="workspace-startup-scan") # P1-3: spawn background idle reaper. Stopped in the lifespan # finally block; awaits the cancellation to avoid a leaked task. start_reaper() @@ -112,4 +116,4 @@ async def handle_jupyter_action(req: JupyterActionRequest) -> dict: status_code=400, detail="'workspace_id' is required for action='get'", ) - return await get_workspace(req.workspace_id) \ No newline at end of file + return await get_workspace(req.workspace_id) diff --git a/runtime/src/runtime/process.py b/runtime/src/runtime/process.py index 5631b23..1ddf853 100644 --- a/runtime/src/runtime/process.py +++ b/runtime/src/runtime/process.py @@ -37,6 +37,7 @@ import asyncio import json import os import secrets +import socket import subprocess import time from typing import TypedDict @@ -163,6 +164,37 @@ def _bump_last_used(record: JupyterProcessRecord) -> None: record["last_used_at"] = time.time() +async def _wait_for_jupyter_ready( + process: subprocess.Popen, + port: int, + *, + timeout_seconds: float = 30, +) -> None: + """Wait until a newly spawned Jupyter server accepts TCP connections. + + Starting the subprocess only means that Python has been forked. Jupyter + still needs several seconds to load extensions and bind its port. The + runtime API must not advertise the workspace as ``running`` before that + happens, otherwise callers immediately receive a transient connection + failure while creating or opening the first notebook. + """ + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError( + f"Jupyter exited during startup (exit code {process.returncode})" + ) + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.2): + return + except OSError: + await asyncio.sleep(0.2) + + raise RuntimeError( + f"Timed out after {timeout_seconds:g}s waiting for Jupyter on port {port}" + ) + + async def start_workspace(ws_id: str) -> dict: async with get_workspace_lock(ws_id): workspace_path = (WORKSPACES_ROOT / ws_id).resolve() @@ -297,6 +329,17 @@ async def start_workspace(ws_id: str) -> dict: detail=f"Failed to start Jupyter: {e}", ) + try: + await _wait_for_jupyter_ready(process, port) + except Exception as exc: + logger.exception(f"Jupyter did not become ready for workspace={ws_id}") + if process.poll() is None: + process.terminate() + raise HTTPException( + status_code=503, + detail=f"Jupyter failed to start: {exc}", + ) from exc + # --------------------------------------------------------- # Save process metadata # ---------------------------------------------------------