feat: 完善模型平台相关功能

This commit is contained in:
Winnie
2026-07-31 19:10:37 +08:00
parent 86988bdaef
commit 49ee2c0a4a
24 changed files with 529 additions and 125 deletions
+20 -7
View File
@@ -1,12 +1,21 @@
COMPOSE_PROJECT_NAME=model-platform COMPOSE_PROJECT_NAME=model-platform-develop
# External port of the Nginx gateway. Only Nginx is exposed to the host # External port of the Nginx gateway. Only Nginx is exposed to the host
# (architecture §2.2); backend/runtime/schedule stay on the Docker internal # (architecture §2.2); backend/runtime/schedule stay on the Docker internal
# network. Override here to expose Nginx on a different host port. # network. Override here to expose Nginx on a different host port.
GATEWAY_PORT=8888 GATEWAY_PORT=8890
# MySQL connection URI (async SQLAlchemy driver) # External MySQL. URL-encode reserved characters in DATABASE_URL.
DATABASE_URL=mysql+asyncmy://model_platform:ChangeMe_MySQL_App_2026@mysql:3306/model_platform?charset=utf8mb4 MYSQL_HOST=127.0.0.1
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=change-me
MYSQL_DATABASE=model_platform
DATABASE_URL=mysql+asyncmy://root:change-me@127.0.0.1:3306/model_platform?charset=utf8mb4
# Demo login is only intended for this self-hosted development UI.
DEMO_AUTH_ENABLED=true
JWT_SECRET=change-this-development-secret
# Object storage (S3-compatible, RustFS). # Object storage (S3-compatible, RustFS).
# RUSTFS_ENDPOINT is the single upstream URL consumed by all 4 services: # RUSTFS_ENDPOINT is the single upstream URL consumed by all 4 services:
@@ -19,8 +28,12 @@ DATABASE_URL=mysql+asyncmy://model_platform:ChangeMe_MySQL_App_2026@mysql:3306/m
# RUSTFS_WORKSPACE_BUCKET — workspace files (notebooks, scripts, working # RUSTFS_WORKSPACE_BUCKET — workspace files (notebooks, scripts, working
# copies); layout is ``s3://<bucket>/<workspace_id>/...``. # copies); layout is ``s3://<bucket>/<workspace_id>/...``.
# Future: RUSTFS_VERSION_BUCKET, RUSTFS_RUN_LOG_BUCKET, ... # Future: RUSTFS_VERSION_BUCKET, RUSTFS_RUN_LOG_BUCKET, ...
RUSTFS_ENDPOINT=http://rustfs:9000 RUSTFS_HOST=127.0.0.1
RUSTFS_ACCESS_KEY=modelplatform RUSTFS_PORT=9000
RUSTFS_SECRET_KEY=ChangeMe_RustFS_2026 RUSTFS_ENDPOINT=http://127.0.0.1:9000
RUSTFS_ACCESS_KEY=change-me
RUSTFS_SECRET_KEY=change-me
RUSTFS_WORKSPACE_BUCKET=workspaces RUSTFS_WORKSPACE_BUCKET=workspaces
RUSTFS_VERSION_BUCKET=versions
RUSTFS_RUN_LOG_BUCKET=run-logs
+4 -1
View File
@@ -1,6 +1,9 @@
FROM python:3.12-slim-bookworm FROM python:3.12-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PYTHONPATH=/app ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/app \
PATH="/app/.venv/bin:${PATH}"
WORKDIR /app WORKDIR /app
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
+72 -1
View File
@@ -12,7 +12,11 @@ from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy import select from sqlalchemy import select
from backend.dependencies import database_session from backend.dependencies import (
RequestContext,
database_session,
request_context,
)
from backend.runtime_client import RuntimeClientError from backend.runtime_client import RuntimeClientError
from common.config import settings from common.config import settings
from common.db.models import Scripts, Users, WorkspaceMembers, Workspaces from common.db.models import Scripts, Users, WorkspaceMembers, Workspaces
@@ -32,6 +36,35 @@ def _b64decode(value: str) -> bytes:
return base64.urlsafe_b64decode(value + padding) return base64.urlsafe_b64decode(value + padding)
def _b64encode(value: bytes) -> str:
return base64.urlsafe_b64encode(value).rstrip(b"=").decode()
def create_jwt_token(user_id: str, *, expires_seconds: int = 3600) -> str:
header = _b64encode(
json.dumps(
{"alg": JWT_ALGORITHM, "typ": "JWT"},
separators=(",", ":"),
).encode()
)
expires_at = int(time.time()) + expires_seconds
payload = _b64encode(
json.dumps(
{"sub": user_id, "exp": expires_at},
separators=(",", ":"),
).encode()
)
signing_input = f"{header}.{payload}".encode()
signature = _b64encode(
hmac.new(
JWT_SECRET.encode(),
signing_input,
hashlib.sha256,
).digest()
)
return f"{header}.{payload}.{signature}"
def verify_jwt_token(token: str) -> dict: def verify_jwt_token(token: str) -> dict:
"""Parse a signed JWT from the access_token cookie. """Parse a signed JWT from the access_token cookie.
@@ -168,6 +201,44 @@ async def load_active_membership(
return row return row
@router.post("/api/v1/auth/demo-session")
async def create_demo_session(
request: Request,
response: Response,
context: RequestContext = Depends(request_context),
) -> dict:
"""Issue the short-lived HttpOnly cookie used by the self-hosted UI.
The normal request-context check still requires an active user/workspace
membership. This endpoint is disabled by default and must be explicitly
enabled by the deployment configuration.
"""
if not settings.demo_auth_enabled:
raise HTTPException(status_code=404, detail="Not Found")
expires_seconds = 3600
response.set_cookie(
"access_token",
create_jwt_token(
context.user.user_id,
expires_seconds=expires_seconds,
),
max_age=expires_seconds,
httponly=True,
secure=request.url.scheme == "https",
samesite="lax",
path="/",
)
return {
"request_id": context.request_id,
"data": {
"user_id": context.user.user_id,
"workspace_id": context.workspace.workspace_id,
"expires_at": int(time.time()) + expires_seconds,
},
"meta": {},
}
@router.get("/api/v1/auth/jupyter") @router.get("/api/v1/auth/jupyter")
async def verify_jupyter_access( async def verify_jupyter_access(
request: Request, request: Request,
+1
View File
@@ -95,6 +95,7 @@ async def complete_resource_upload(
upload_id, upload_id,
{ {
"usage_type": "data_resource", "usage_type": "data_resource",
"file_name": payload.resource_name,
"visibility": payload.visibility, "visibility": payload.visibility,
"is_immutable": False, "is_immutable": False,
}, },
+11
View File
@@ -130,12 +130,20 @@ def script_payload(
) -> dict[str, Any]: ) -> dict[str, Any]:
if isinstance(storage_object, dict): if isinstance(storage_object, dict):
relative_path = storage_object.get("relative_path") relative_path = storage_object.get("relative_path")
object_key = storage_object.get("object_key")
content_hash = storage_object.get("content_hash") content_hash = storage_object.get("content_hash")
size_bytes = storage_object.get("size_bytes", 0) size_bytes = storage_object.get("size_bytes", 0)
else: else:
relative_path = storage_object.relative_path relative_path = storage_object.relative_path
object_key = storage_object.object_key
content_hash = storage_object.content_hash content_hash = storage_object.content_hash
size_bytes = storage_object.size_bytes size_bytes = storage_object.size_bytes
workspace_prefix = f"{script.workspace_id}/"
jupyter_path = (
object_key[len(workspace_prefix):]
if object_key and object_key.startswith(workspace_prefix)
else object_key
)
return { return {
"script_id": script.script_id, "script_id": script.script_id,
"workspace_id": script.workspace_id, "workspace_id": script.workspace_id,
@@ -146,6 +154,7 @@ def script_payload(
"visibility": script.visibility, "visibility": script.visibility,
"status": script.status, "status": script.status,
"relative_path": relative_path, "relative_path": relative_path,
"jupyter_path": jupyter_path,
"content_hash": content_hash, "content_hash": content_hash,
"size_bytes": size_bytes, "size_bytes": size_bytes,
"created_at": script.created_at.isoformat(), "created_at": script.created_at.isoformat(),
@@ -299,6 +308,7 @@ async def create_script_record(
visibility=visibility, visibility=visibility,
is_immutable=False, is_immutable=False,
idempotency_key=f"script:{context.workspace.workspace_id}:{relative_path}", idempotency_key=f"script:{context.workspace.workspace_id}:{relative_path}",
relative_path=relative_path,
) )
object_id = storage_data["storage_object_id"] object_id = storage_data["storage_object_id"]
script = await session.scalar( script = await session.scalar(
@@ -653,6 +663,7 @@ async def update_script(
f"script-update:{script.script_id}:" f"script-update:{script.script_id}:"
f"{hashlib.sha256(content).hexdigest()}" f"{hashlib.sha256(content).hexdigest()}"
), ),
relative_path=storage_object.relative_path,
) )
script.current_object_id = storage_data["storage_object_id"] script.current_object_id = storage_data["storage_object_id"]
script.updated_at = datetime.now(UTC).replace(tzinfo=None) script.updated_at = datetime.now(UTC).replace(tzinfo=None)
+26 -7
View File
@@ -99,6 +99,7 @@ def storage_payload(item: StorageObjects) -> dict[str, Any]:
"storage_backend": item.storage_backend, "storage_backend": item.storage_backend,
"usage_type": item.usage_type, "usage_type": item.usage_type,
"storage_uri": item.storage_uri, "storage_uri": item.storage_uri,
"object_key": item.object_key,
"relative_path": item.relative_path, "relative_path": item.relative_path,
"file_name": item.file_name, "file_name": item.file_name,
"file_extension": item.file_extension, "file_extension": item.file_extension,
@@ -199,12 +200,13 @@ async def create_upload_record(
else: else:
upload_id = new_ulid() upload_id = new_ulid()
bucket_name = resolve_bucket(payload.usage_type, workspace=workspace) bucket_name = resolve_bucket(payload.usage_type, workspace=workspace)
# Object key is a flat two-level path: workspace id + upload id. The # Keep the opaque upload id while preserving the original extension.
# original file name and content type live in the StorageObjects row # Jupyter selects its editor from this suffix, so an extensionless
# (file_name / mime_type / object_key) — they are not part of the # object would make notebooks look like generic JSON/text files.
# key itself, so the bucket can be reorganised without rewriting file_extension = PurePosixPath(
# the database. safe_file_name(payload.file_name)
object_key = f"{payload.workspace_id}/{upload_id}" ).suffix.lower()
object_key = f"{payload.workspace_id}/{upload_id}{file_extension}"
upload = UploadSessions( upload = UploadSessions(
upload_id=upload_id, upload_id=upload_id,
workspace_id=payload.workspace_id, workspace_id=payload.workspace_id,
@@ -465,7 +467,21 @@ async def create_server_object(
session, session,
request) request)
if upload_result.get("status") == "completed": if upload_result.get("status") == "completed":
return {"data": upload_result["storage_object"], "meta": {"reused": True}} existing_data = upload_result["storage_object"]
if (
payload.relative_path
and existing_data
and existing_data.get("relative_path") != payload.relative_path
):
existing_item = await session.get(
StorageObjects,
existing_data["storage_object_id"],
)
if existing_item is not None:
existing_item.relative_path = payload.relative_path
await session.flush()
existing_data = storage_payload(existing_item)
return {"data": existing_data, "meta": {"reused": True}}
upload = await session.get(UploadSessions, upload_result["upload_id"]) upload = await session.get(UploadSessions, upload_result["upload_id"])
if upload is None: if upload is None:
@@ -483,10 +499,13 @@ async def create_server_object(
upload.upload_id, upload.upload_id,
CompleteUploadRequest( CompleteUploadRequest(
usage_type=payload.usage_type, usage_type=payload.usage_type,
file_name=payload.file_name,
visibility=payload.visibility, visibility=payload.visibility,
is_immutable=payload.is_immutable), is_immutable=payload.is_immutable),
session, session,
request) request)
item.relative_path = payload.relative_path
await session.flush()
return {"data": storage_payload(item), "meta": {"reused": False}} return {"data": storage_payload(item), "meta": {"reused": False}}
+5 -1
View File
@@ -38,6 +38,10 @@ class Settings(BaseSettings):
default="dev-only-not-for-production", default="dev-only-not-for-production",
description="HS256 secret used by backend's jupyter auth_request.", description="HS256 secret used by backend's jupyter auth_request.",
) )
demo_auth_enabled: bool = Field(
default=False,
description="Enable the self-hosted UI's short-lived demo session cookie.",
)
# ── service identity ────────────────────────────────────────── # ── service identity ──────────────────────────────────────────
service_name: str = Field( service_name: str = Field(
@@ -148,4 +152,4 @@ def get_settings() -> Settings:
settings: Settings = get_settings() settings: Settings = get_settings()
__all__ = ["Settings", "get_settings", "settings"] __all__ = ["Settings", "get_settings", "settings"]
+2
View File
@@ -100,6 +100,7 @@ class StorageClient:
visibility: str, visibility: str,
is_immutable: bool, is_immutable: bool,
idempotency_key: str, idempotency_key: str,
relative_path: str | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
result = await self._request( result = await self._request(
"POST", "POST",
@@ -114,6 +115,7 @@ class StorageClient:
"visibility": visibility, "visibility": visibility,
"is_immutable": is_immutable, "is_immutable": is_immutable,
"idempotency_key": idempotency_key, "idempotency_key": idempotency_key,
"relative_path": relative_path,
}, },
) )
return result["data"] return result["data"]
+2
View File
@@ -56,6 +56,7 @@ class CompleteUploadRequest(StrictModel):
"working_copy", "working_copy",
"public_script", "public_script",
] ]
file_name: str = Field(min_length=1, max_length=255)
visibility: Literal["private", "workspace", "public"] = "private" visibility: Literal["private", "workspace", "public"] = "private"
is_immutable: bool = False is_immutable: bool = False
@@ -78,6 +79,7 @@ class ServerObjectRequest(StrictModel):
visibility: Literal["private", "workspace", "public"] = "private" visibility: Literal["private", "workspace", "public"] = "private"
is_immutable: bool = False is_immutable: bool = False
idempotency_key: str = Field(min_length=8, max_length=128) idempotency_key: str = Field(min_length=8, max_length=128)
relative_path: str | None = Field(default=None, max_length=1024)
class DownloadUrlRequest(StrictModel): class DownloadUrlRequest(StrictModel):
+3 -1
View File
@@ -39,6 +39,8 @@ server {
location /api/ { location /api/ {
proxy_pass http://backend:8000/api/; proxy_pass http://backend:8000/api/;
proxy_set_header Host $host; proxy_set_header Host $host;
proxy_set_header X-Forwarded-Host $http_host;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
} }
@@ -127,4 +129,4 @@ server {
location /jupyter/ { location /jupyter/ {
return 403 "Access Denied"; return 403 "Access Denied";
} }
} }
+83 -31
View File
@@ -1,6 +1,21 @@
version: '3.8'
services: services:
migrate:
build:
context: .
dockerfile: backend/Dockerfile
restart: "no"
command:
- uv
- run
- --frozen
- --package
- backend
- alembic
- upgrade
- head
environment:
DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required}
web: web:
build: build:
context: . context: .
@@ -12,14 +27,19 @@ services:
# /etc/nginx/conf.d/default.conf before exec'ing nginx. # /etc/nginx/conf.d/default.conf before exec'ing nginx.
ports: ports:
- "${GATEWAY_PORT:-8888}:80" - "${GATEWAY_PORT:-8888}:80"
volumes:
- ./default.conf:/etc/nginx/conf.d/default.conf.template:ro
- ./scripts/nginx-entrypoint.sh:/docker-entrypoint.sh:ro
environment: environment:
RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:-http://rustfs:9000} RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:?RUSTFS_ENDPOINT is required}
depends_on: depends_on:
- backend backend:
- runtime condition: service_healthy
runtime:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://127.0.0.1/ >/dev/null"]
interval: 10s
timeout: 3s
retries: 12
start_period: 10s
backend: backend:
build: build:
@@ -29,17 +49,29 @@ services:
# No host port: architecture §2.2 — only Nginx is externally reachable. # No host port: architecture §2.2 — only Nginx is externally reachable.
# No local-FS volume: backend stores everything in RustFS (RUSTFS_*). # No local-FS volume: backend stores everything in RustFS (RUSTFS_*).
environment: environment:
DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required}
JWT_SECRET: ${JWT_SECRET:-local-jwt-secret} SERVICE_NAME: model-platform-backend
JWT_SECRET: ${JWT_SECRET:?JWT_SECRET is required}
DEMO_AUTH_ENABLED: ${DEMO_AUTH_ENABLED:-false}
RUNTIME_API_URL: http://runtime:8000 RUNTIME_API_URL: http://runtime:8000
RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:-http://rustfs:9000} RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:?RUSTFS_ENDPOINT is required}
RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:-modelplatform} RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:?RUSTFS_ACCESS_KEY is required}
RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:?RUSTFS_SECRET_KEY is required}
volumes: RUSTFS_WORKSPACE_BUCKET: ${RUSTFS_WORKSPACE_BUCKET:-workspaces}
- ./backend:/app/backend:ro RUSTFS_VERSION_BUCKET: ${RUSTFS_VERSION_BUCKET:-versions}
- ./common:/app/common:ro RUSTFS_RUN_LOG_BUCKET: ${RUSTFS_RUN_LOG_BUCKET:-run-logs}
READINESS_TARGETS: ${MYSQL_HOST:?MYSQL_HOST is required}:${MYSQL_PORT:-3306},${RUSTFS_HOST:?RUSTFS_HOST is required}:${RUSTFS_PORT:-9000},runtime:8000
depends_on: depends_on:
- runtime migrate:
condition: service_completed_successfully
runtime:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "curl -fsS http://127.0.0.1:8000/health/ready >/dev/null"]
interval: 10s
timeout: 5s
retries: 18
start_period: 20s
runtime: runtime:
build: build:
@@ -54,23 +86,30 @@ services:
- apparmor:unconfined - apparmor:unconfined
# No host port: architecture §2.2 — only Nginx is externally reachable. # No host port: architecture §2.2 — only Nginx is externally reachable.
environment: environment:
DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required}
SERVICE_NAME: runtime-manager SERVICE_NAME: runtime-manager
WORKSPACES_ROOT: /workspace/workspaces WORKSPACES_ROOT: /workspace/workspaces
PUBLIC_BASE_URL: http://runtime PUBLIC_BASE_URL: http://runtime
REMOTE_BUCKET: rustfs:workspaces REMOTE_BUCKET: rustfs:${RUSTFS_WORKSPACE_BUCKET:-workspaces}
RCLONE_CONFIG_RUSTFS_TYPE: s3 RCLONE_CONFIG_RUSTFS_TYPE: s3
RCLONE_CONFIG_RUSTFS_PROVIDER: Other RCLONE_CONFIG_RUSTFS_PROVIDER: Other
RCLONE_CONFIG_RUSTFS_ACCESS_KEY_ID: ${RUSTFS_ACCESS_KEY:-modelplatform} RCLONE_CONFIG_RUSTFS_ACCESS_KEY_ID: ${RUSTFS_ACCESS_KEY:?RUSTFS_ACCESS_KEY is required}
RCLONE_CONFIG_RUSTFS_SECRET_ACCESS_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} RCLONE_CONFIG_RUSTFS_SECRET_ACCESS_KEY: ${RUSTFS_SECRET_KEY:?RUSTFS_SECRET_KEY is required}
RCLONE_CONFIG_RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:-http://rustfs:9000} RCLONE_CONFIG_RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:?RUSTFS_ENDPOINT is required}
RCLONE_CONFIG_RUSTFS_ENV_AUTH: "false" RCLONE_CONFIG_RUSTFS_ENV_AUTH: "false"
RCLONE_CONFIG_RUSTFS_FORCE_PATH_STYLE: "true" RCLONE_CONFIG_RUSTFS_FORCE_PATH_STYLE: "true"
RCLONE_CONFIG_RUSTFS_REGION: other RCLONE_CONFIG_RUSTFS_REGION: other
volumes: depends_on:
- ./deploy/data/workspaces:/workspace/workspaces migrate:
- ./runtime:/app/runtime:ro condition: service_completed_successfully
- ./common:/app/common:ro healthcheck:
test:
- CMD-SHELL
- grep -q ' /workspace/workspaces .* - fuse.rclone ' /proc/self/mountinfo && curl -fsS http://127.0.0.1:8000/api/v1/health >/dev/null
interval: 10s
timeout: 5s
retries: 18
start_period: 30s
schedule: schedule:
build: build:
@@ -81,9 +120,22 @@ services:
# No local-FS volume: schedule executes nodes via tempfile.TemporaryDirectory # No local-FS volume: schedule executes nodes via tempfile.TemporaryDirectory
# under Python's default temp dir (cleaned per-run); artifacts live in RustFS. # under Python's default temp dir (cleaned per-run); artifacts live in RustFS.
environment: environment:
DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required}
RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:-http://rustfs:9000} SERVICE_NAME: schedule-executor
RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:-modelplatform} BACKEND_API_URL: http://backend:8000
RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:?RUSTFS_ENDPOINT is required}
RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:?RUSTFS_ACCESS_KEY is required}
RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:?RUSTFS_SECRET_KEY is required}
RUSTFS_WORKSPACE_BUCKET: ${RUSTFS_WORKSPACE_BUCKET:-workspaces}
RUSTFS_VERSION_BUCKET: ${RUSTFS_VERSION_BUCKET:-versions}
RUSTFS_RUN_LOG_BUCKET: ${RUSTFS_RUN_LOG_BUCKET:-run-logs}
READINESS_TARGETS: ${MYSQL_HOST:?MYSQL_HOST is required}:${MYSQL_PORT:-3306},${RUSTFS_HOST:?RUSTFS_HOST is required}:${RUSTFS_PORT:-9000},backend:8000
depends_on: depends_on:
- backend backend:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "curl -fsS http://127.0.0.1:8000/health/ready >/dev/null"]
interval: 10s
timeout: 5s
retries: 18
start_period: 20s
+7 -2
View File
@@ -4,9 +4,14 @@ RUN corepack enable && corepack prepare pnpm@10.15.1 --activate
COPY frontend/package.json frontend/pnpm-lock.yaml ./ COPY frontend/package.json frontend/pnpm-lock.yaml ./
RUN pnpm install --frozen-lockfile RUN pnpm install --frozen-lockfile
COPY frontend/ ./ COPY frontend/ ./
RUN pnpm build RUN pnpm typecheck && pnpm build
FROM nginx:alpine FROM nginx:alpine
COPY ./default.conf /etc/nginx/conf.d/default.conf COPY ./default.conf /etc/nginx/conf.d/default.conf.template
COPY ./scripts/nginx-entrypoint.sh /usr/local/bin/model-platform-entrypoint.sh
RUN sed -i 's/\r$//' /usr/local/bin/model-platform-entrypoint.sh \
&& chmod +x /usr/local/bin/model-platform-entrypoint.sh
COPY --from=frontend-build /app/build/client /usr/share/nginx/html COPY --from=frontend-build /app/build/client /usr/share/nginx/html
EXPOSE 80 EXPOSE 80
ENTRYPOINT ["/usr/local/bin/model-platform-entrypoint.sh"]
CMD ["nginx", "-g", "daemon off;"]
+1 -1
View File
@@ -1,4 +1,4 @@
import { FormEvent, useEffect, useState } from "react"; import { type FormEvent, useEffect, useState } from "react";
import { import {
ApiRequestError, ApiRequestError,
@@ -1,6 +1,6 @@
import { import {
type ChangeEvent, type ChangeEvent,
FormEvent, type FormEvent,
type MouseEvent as ReactMouseEvent, type MouseEvent as ReactMouseEvent,
useEffect, useEffect,
useMemo, useMemo,
@@ -1,8 +1,8 @@
import { import {
DragEvent, type DragEvent,
FormEvent, type FormEvent,
MouseEvent as ReactMouseEvent, type MouseEvent as ReactMouseEvent,
PointerEvent as ReactPointerEvent, type PointerEvent as ReactPointerEvent,
useEffect, useEffect,
useMemo, useMemo,
useRef, useRef,
+46 -45
View File
@@ -87,6 +87,7 @@ export type ScriptItem = {
visibility: Visibility; visibility: Visibility;
status: string; status: string;
relative_path: string; relative_path: string;
jupyter_path: string;
content_hash: string; content_hash: string;
size_bytes: number; size_bytes: number;
created_at: string; created_at: string;
@@ -195,16 +196,18 @@ function initialContent(scriptType: ScriptType): string {
{ {
cells: [ cells: [
{ {
id: "intro",
cell_type: "markdown", cell_type: "markdown",
metadata: {}, metadata: {},
source: ["# 新建模型实验\\n", "在这里开始数据探索与模型构建。"], source: ["# 新建模型实验\n", "在这里开始数据探索与模型构建。"],
}, },
{ {
id: "main",
cell_type: "code", cell_type: "code",
execution_count: null, execution_count: null,
metadata: {}, metadata: {},
outputs: [], outputs: [],
source: ["print('Hello, Model Platform!')\\n"], source: ["print('Hello, Model Platform!')\n"],
}, },
], ],
metadata: { metadata: {
@@ -325,6 +328,7 @@ export type FileLockSession = {
export type ActiveEditSession = FileLockSession & { export type ActiveEditSession = FileLockSession & {
script_id: string; script_id: string;
script_name: string; script_name: string;
jupyter_path: string;
lock_token: string; lock_token: string;
ticket_expires_at?: string; ticket_expires_at?: string;
}; };
@@ -356,70 +360,67 @@ export type StableVersion = {
export async function acquireFileLock( export async function acquireFileLock(
script: ScriptItem, script: ScriptItem,
): Promise<ActiveEditSession> { ): Promise<ActiveEditSession> {
const session = await apiRequest<FileLockSession>( const now = Date.now();
`/api/v1/files/${script.current_object_id}/lock`,
{ method: "POST" },
);
if (!session.lock_token) {
throw new Error("加锁成功响应缺少 lock_token");
}
return { return {
...session, edit_session_id: crypto.randomUUID().replaceAll("-", ""),
workspace_id: script.workspace_id,
storage_object_id: script.current_object_id,
user_id: demoContext.userId,
session_status: "active",
lease_seconds: 3600,
heartbeat_interval_seconds: 300,
expires_at: new Date(now + 3600_000).toISOString(),
runtime_id: script.workspace_id,
jupyter_session_id: "demo-session",
relative_path: script.relative_path,
lock_token: "demo-unlocked-session",
script_id: script.script_id, script_id: script.script_id,
script_name: script.script_name, script_name: script.script_name,
lock_token: session.lock_token, jupyter_path: script.jupyter_path,
}; };
} }
export async function heartbeatFileLock( export async function heartbeatFileLock(
session: ActiveEditSession, session: ActiveEditSession,
): Promise<FileLockSession> { ): Promise<FileLockSession> {
return apiRequest<FileLockSession>( return {
`/api/v1/file-locks/${session.edit_session_id}/heartbeat`, ...session,
{ expires_at: new Date(Date.now() + 3600_000).toISOString(),
method: "POST", };
body: JSON.stringify({ lock_token: session.lock_token }),
},
);
} }
export async function releaseFileLock( export async function releaseFileLock(
session: ActiveEditSession, session: ActiveEditSession,
): Promise<FileLockSession> { ): Promise<FileLockSession> {
return apiRequest<FileLockSession>( return { ...session, session_status: "closed" };
`/api/v1/file-locks/${session.edit_session_id}`,
{
method: "DELETE",
body: JSON.stringify({ lock_token: session.lock_token }),
},
);
} }
export function releaseFileLockOnUnload(session: ActiveEditSession): void { export function releaseFileLockOnUnload(_session: ActiveEditSession): void {
void fetch(`/api/v1/file-locks/${session.edit_session_id}`, { // The current Backend deliberately has no persisted file-lock API.
method: "DELETE",
credentials: "same-origin",
keepalive: true,
headers: {
"Content-Type": "application/json",
"X-User-ID": demoContext.userId,
"X-Workspace-ID": demoContext.workspaceId,
"X-Request-ID": crypto.randomUUID().replaceAll("-", ""),
},
body: JSON.stringify({ lock_token: session.lock_token }),
});
} }
export async function createJupyterAccessTicket( export async function createJupyterAccessTicket(
session: ActiveEditSession, session: ActiveEditSession,
): Promise<JupyterAccessTicket> { ): Promise<JupyterAccessTicket> {
return apiRequest<JupyterAccessTicket>("/api/v1/jupyter/access-tickets", { const result = await apiRequest<{ expires_at: number }>(
method: "POST", "/api/v1/auth/demo-session",
body: JSON.stringify({ {
edit_session_id: session.edit_session_id, method: "POST",
lock_token: session.lock_token, },
}), );
}); const editorRoute = session.script_name.toLowerCase().endsWith(".ipynb")
? "notebooks"
: "edit";
const encodedPath = session.jupyter_path
.split("/")
.filter(Boolean)
.map(encodeURIComponent)
.join("/");
return {
edit_session_id: session.edit_session_id,
jupyter_url: `/jupyter/${encodeURIComponent(session.workspace_id)}/${editorRoute}/${encodedPath}`,
expires_at: new Date(result.expires_at * 1000).toISOString(),
};
} }
export async function listScriptVersions( export async function listScriptVersions(
+2 -2
View File
@@ -8,11 +8,11 @@ export default defineConfig({
port: 5173, port: 5173,
proxy: { proxy: {
"/api": { "/api": {
target: process.env.VITE_GATEWAY_URL || "http://127.0.0.1:8081", target: process.env.VITE_GATEWAY_URL || "http://127.0.0.1:8890",
changeOrigin: true, changeOrigin: true,
}, },
"/jupyter": { "/jupyter": {
target: process.env.VITE_GATEWAY_URL || "http://127.0.0.1:8081", target: process.env.VITE_GATEWAY_URL || "http://127.0.0.1:8890",
changeOrigin: true, changeOrigin: true,
ws: true, ws: true,
}, },
@@ -0,0 +1,168 @@
"""seed the self-hosted demo users and workspaces
Revision ID: b71c4f2a9d10
Revises: 8d86e2f82860
Create Date: 2026-07-31 16:00:00
"""
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
revision: str = "b71c4f2a9d10"
down_revision: str | Sequence[str] | None = "8d86e2f82860"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
ADMIN_ROLE_ID = "0000000000000000000000000A"
DEVELOPER_ROLE_ID = "0000000000000000000000000B"
USERS = (
("0000000000RF6FG1SDBXG59S13", "admin-zhang", "张三", ADMIN_ROLE_ID),
("0000000000H2QYCGPCWQM1JSGS", "admin-li", "李四", ADMIN_ROLE_ID),
("0000000000RWG40ESZPGJT629J", "dev-wang", "王五", DEVELOPER_ROLE_ID),
("00000000004CQV7WASJA6N6FW4", "dev-zhao", "赵六", DEVELOPER_ROLE_ID),
)
WORKSPACES = (
("00000000000BM630VT9ARVFZPC", "model-development", "模型开发 Workspace"),
("0000000000AE0NC0V5T424KK86", "risk-validation", "风险验证 Workspace"),
)
def upgrade() -> None:
roles = sa.table(
"roles",
sa.column("role_id", sa.String),
sa.column("role_code", sa.String),
sa.column("role_name", sa.String),
sa.column("role_scope", sa.String),
sa.column("is_builtin", sa.Integer),
sa.column("description", sa.String),
)
users = sa.table(
"users",
sa.column("user_id", sa.String),
sa.column("username", sa.String),
sa.column("display_name", sa.String),
sa.column("password_hash", sa.String),
sa.column("status", sa.String),
sa.column("email", sa.String),
sa.column("platform_role_id", sa.String),
)
workspaces = sa.table(
"workspaces",
sa.column("workspace_id", sa.String),
sa.column("workspace_code", sa.String),
sa.column("workspace_name", sa.String),
sa.column("active_root_uri", sa.String),
sa.column("status", sa.String),
sa.column("created_by", sa.String),
sa.column("description", sa.String),
)
members = sa.table(
"workspace_members",
sa.column("workspace_id", sa.String),
sa.column("user_id", sa.String),
sa.column("role_id", sa.String),
sa.column("member_status", sa.String),
)
op.bulk_insert(
roles,
[
{
"role_id": ADMIN_ROLE_ID,
"role_code": "admin",
"role_name": "管理员",
"role_scope": "workspace",
"is_builtin": 1,
"description": "Self-hosted workspace administrator",
},
{
"role_id": DEVELOPER_ROLE_ID,
"role_code": "developer",
"role_name": "开发人员",
"role_scope": "workspace",
"is_builtin": 1,
"description": "Self-hosted workspace developer",
},
],
)
op.bulk_insert(
users,
[
{
"user_id": user_id,
"username": username,
"display_name": display_name,
"password_hash": "demo-login-disabled",
"status": "active",
"email": f"{username}@model-platform.local",
"platform_role_id": role_id,
}
for user_id, username, display_name, role_id in USERS
],
)
op.bulk_insert(
workspaces,
[
{
"workspace_id": workspace_id,
"workspace_code": workspace_code,
"workspace_name": workspace_name,
"active_root_uri": f"s3://workspaces/{workspace_id}/",
"status": "active",
"created_by": USERS[0][0],
"description": "Self-hosted demo workspace",
}
for workspace_id, workspace_code, workspace_name in WORKSPACES
],
)
op.bulk_insert(
members,
[
{
"workspace_id": workspace_id,
"user_id": user_id,
"role_id": role_id,
"member_status": "active",
}
for workspace_id, _, _ in WORKSPACES
for user_id, _, _, role_id in USERS
],
)
def downgrade() -> None:
connection = op.get_bind()
workspace_ids = [workspace_id for workspace_id, _, _ in WORKSPACES]
user_ids = [user_id for user_id, _, _, _ in USERS]
connection.execute(
sa.text(
"DELETE FROM workspace_members "
"WHERE workspace_id IN :workspace_ids AND user_id IN :user_ids"
).bindparams(
sa.bindparam("workspace_ids", expanding=True),
sa.bindparam("user_ids", expanding=True),
),
{"workspace_ids": workspace_ids, "user_ids": user_ids},
)
connection.execute(
sa.text("DELETE FROM workspaces WHERE workspace_id IN :workspace_ids")
.bindparams(sa.bindparam("workspace_ids", expanding=True)),
{"workspace_ids": workspace_ids},
)
connection.execute(
sa.text("DELETE FROM users WHERE user_id IN :user_ids")
.bindparams(sa.bindparam("user_ids", expanding=True)),
{"user_ids": user_ids},
)
connection.execute(
sa.text("DELETE FROM roles WHERE role_id IN :role_ids")
.bindparams(sa.bindparam("role_ids", expanding=True)),
{"role_ids": [ADMIN_ROLE_ID, DEVELOPER_ROLE_ID]},
)
+4 -1
View File
@@ -1,6 +1,9 @@
FROM python:3.12-slim-bookworm FROM python:3.12-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PYTHONPATH=/app ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/app \
PATH="/app/.venv/bin:${PATH}"
WORKDIR /app WORKDIR /app
# 从官方 Rclone 镜像直接复制 rclone 二进制文件(高效、稳定) # 从官方 Rclone 镜像直接复制 rclone 二进制文件(高效、稳定)
COPY --from=rclone/rclone:latest /usr/local/bin/rclone /usr/local/bin/rclone COPY --from=rclone/rclone:latest /usr/local/bin/rclone /usr/local/bin/rclone
+27 -11
View File
@@ -21,18 +21,34 @@ REMOTE_BUCKET = settings.remote_bucket
RCLONE_PROCESS: subprocess.Popen | None = None RCLONE_PROCESS: subprocess.Popen | None = None
def is_mountpoint(path: Path) -> bool: def is_rclone_mount(path: Path) -> bool:
result = subprocess.run( """Return True only for the FUSE mount created by this service.
["mountpoint", "-q", str(path)],
stdout=subprocess.DEVNULL, Docker volumes and bind mounts are mount points too, so ``mountpoint -q``
stderr=subprocess.DEVNULL, cannot distinguish the writable mount target from an active rclone mount.
) Linux exposes the filesystem type after `` - `` in ``mountinfo``.
return result.returncode == 0 """
target = str(path.resolve())
try:
lines = Path("/proc/self/mountinfo").read_text().splitlines()
except OSError:
return False
for line in lines:
fields = line.split()
if len(fields) < 10 or fields[4].replace("\\040", " ") != target:
continue
try:
separator = fields.index("-")
except ValueError:
continue
if separator + 1 < len(fields) and fields[separator + 1] == "fuse.rclone":
return True
return False
def start_rclone_mount() -> None: def start_rclone_mount() -> None:
global RCLONE_PROCESS global RCLONE_PROCESS
if is_mountpoint(WORKSPACES_ROOT): if is_rclone_mount(WORKSPACES_ROOT):
logger.info(f"Mountpoint already exists: {WORKSPACES_ROOT}") logger.info(f"Mountpoint already exists: {WORKSPACES_ROOT}")
return return
@@ -62,7 +78,7 @@ def start_rclone_mount() -> None:
timeout = 20 timeout = 20
while timeout > 0: while timeout > 0:
if is_mountpoint(WORKSPACES_ROOT): if is_rclone_mount(WORKSPACES_ROOT):
logger.info(f"rclone mount ready: {WORKSPACES_ROOT}") logger.info(f"rclone mount ready: {WORKSPACES_ROOT}")
return return
if RCLONE_PROCESS.poll() is not None: if RCLONE_PROCESS.poll() is not None:
@@ -85,10 +101,10 @@ def stop_rclone_mount() -> None:
logger.warning("Force killing rclone") logger.warning("Force killing rclone")
RCLONE_PROCESS.kill() RCLONE_PROCESS.kill()
if is_mountpoint(WORKSPACES_ROOT): if is_rclone_mount(WORKSPACES_ROOT):
logger.info(f"Unmount {WORKSPACES_ROOT}") logger.info(f"Unmount {WORKSPACES_ROOT}")
result = subprocess.run(["fusermount3", "-u", str(WORKSPACES_ROOT)]) result = subprocess.run(["fusermount3", "-u", str(WORKSPACES_ROOT)])
if result.returncode != 0: if result.returncode != 0:
subprocess.run(["umount", "-l", str(WORKSPACES_ROOT)]) subprocess.run(["umount", "-l", str(WORKSPACES_ROOT)])
logger.info("rclone stopped") logger.info("rclone stopped")
+9 -3
View File
@@ -13,6 +13,7 @@ import os
import secrets import secrets
import subprocess import subprocess
import time import time
from pathlib import Path
from typing import TypedDict from typing import TypedDict
from fastapi import HTTPException from fastapi import HTTPException
@@ -23,6 +24,7 @@ from common.utils import get_free_port, start_process
from runtime.mount import WORKSPACES_ROOT from runtime.mount import WORKSPACES_ROOT
PUBLIC_BASE_URL = settings.public_base_url PUBLIC_BASE_URL = settings.public_base_url
JUPYTER_PROCESS_CWD = Path("/app")
class JupyterProcessRecord(TypedDict): class JupyterProcessRecord(TypedDict):
@@ -106,7 +108,7 @@ async def start_workspace(ws_id: str) -> dict:
"--allow-root", "--allow-root",
f"--ServerApp.token={token}", f"--ServerApp.token={token}",
f"--ServerApp.base_url={base_path}", f"--ServerApp.base_url={base_path}",
"--notebook-dir=.", f"--ServerApp.root_dir={workspace_path}",
"--ServerApp.terminals_enabled=False", "--ServerApp.terminals_enabled=False",
"--NotebookApp.terminals_enabled=False", "--NotebookApp.terminals_enabled=False",
"--ServerApp.allow_origin=*", "--ServerApp.allow_origin=*",
@@ -116,7 +118,11 @@ async def start_workspace(ws_id: str) -> dict:
] ]
try: try:
process, log_file = start_process(cmd, workspace_path) # Keep the server process cwd off the rclone/FUSE mount. Remote
# S3 directory refreshes can replace a virtual directory inode,
# leaving a long-lived cwd marked "(deleted)" and making
# os.getcwd() fail while Jupyter starts kernels or nbconvert.
process, log_file = start_process(cmd, JUPYTER_PROCESS_CWD)
except Exception as e: except Exception as e:
logger.error(f"Failed to start Jupyter for {ws_id}: {e}") logger.error(f"Failed to start Jupyter for {ws_id}: {e}")
raise HTTPException( raise HTTPException(
@@ -264,4 +270,4 @@ async def scan_workspaces() -> None:
except Exception as err: except Exception as err:
logger.error(f"Startup failed for workspace '{entry}': {err}") logger.error(f"Startup failed for workspace '{entry}': {err}")
await asyncio.gather(*[_start(entry) for entry in entries]) await asyncio.gather(*[_start(entry) for entry in entries])
+5 -2
View File
@@ -1,6 +1,9 @@
FROM python:3.12-slim-bookworm FROM python:3.12-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PYTHONPATH=/app ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/app \
PATH="/app/.venv/bin:${PATH}"
WORKDIR /app WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \ RUN apt-get update && apt-get install -y --no-install-recommends \
@@ -10,7 +13,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
g++ \ g++ \
python3-dev \ python3-dev \
build-essential \ build-essential \
&& rm -rf /var/lib/apt/lists/* \ && rm -rf /var/lib/apt/lists/*
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
COPY pyproject.toml uv.lock ./ COPY pyproject.toml uv.lock ./
+24 -2
View File
@@ -33,6 +33,20 @@ from schedule.context import naive_utc
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
# APScheduler's persistent SQLAlchemy job store pickles each job. A bound
# ``CronScheduler`` method captures this instance (including SQLAlchemy engine
# state) and therefore cannot be pickled. Keep the persisted callable at
# module scope and resolve the process-local callback when the job fires.
_ACTIVE_TRIGGER: Callable[[str], Awaitable[None]] | None = None
async def dispatch_persisted_cron(schedule_id: str) -> None:
"""Dispatch one persisted cron tick through the active service."""
callback = _ACTIVE_TRIGGER
if callback is None:
raise RuntimeError("cron trigger callback is not initialized")
await callback(schedule_id)
class CronScheduler: class CronScheduler:
"""Manages cron triggers in APScheduler, backed by a MySQL jobstore.""" """Manages cron triggers in APScheduler, backed by a MySQL jobstore."""
@@ -44,12 +58,15 @@ class CronScheduler:
database_url: str, database_url: str,
on_trigger: Callable[[str], Awaitable[None]], on_trigger: Callable[[str], Awaitable[None]],
) -> None: ) -> None:
global _ACTIVE_TRIGGER
self.session_factory = session_factory self.session_factory = session_factory
self.scheduler = AsyncIOScheduler( self.scheduler = AsyncIOScheduler(
jobstores={"default": build_sqlalchemy_jobstore(database_url)}, jobstores={"default": build_sqlalchemy_jobstore(database_url)},
timezone=UTC, timezone=UTC,
) )
self._on_trigger = on_trigger self._on_trigger = on_trigger
_ACTIVE_TRIGGER = on_trigger
self._sync_task: asyncio.Task[None] | None = None self._sync_task: asyncio.Task[None] | None = None
def start(self) -> None: def start(self) -> None:
@@ -62,6 +79,8 @@ class CronScheduler:
async def close(self) -> None: async def close(self) -> None:
"""Cancel the sync loop and shut APScheduler down.""" """Cancel the sync loop and shut APScheduler down."""
global _ACTIVE_TRIGGER
if self._sync_task is not None: if self._sync_task is not None:
self._sync_task.cancel() self._sync_task.cancel()
try: try:
@@ -71,11 +90,14 @@ class CronScheduler:
self._sync_task = None self._sync_task = None
if self.scheduler.running: if self.scheduler.running:
self.scheduler.shutdown(wait=False) self.scheduler.shutdown(wait=False)
if _ACTIVE_TRIGGER is self._on_trigger:
_ACTIVE_TRIGGER = None
async def trigger(self, schedule_id: str) -> None: async def trigger(self, schedule_id: str) -> None:
"""APScheduler cron tick callback. """APScheduler cron tick callback.
Wired via ``add_job(self.trigger, args=[schedule_id], ...)``. The The persisted job calls :func:`dispatch_persisted_cron`, which then
resolves this process-local callback. The
standard on_trigger is ``SchedulerService.trigger_schedule`` which standard on_trigger is ``SchedulerService.trigger_schedule`` which
posts back to Backend; Backend then writes the Outbox row that the posts back to Backend; Backend then writes the Outbox row that the
orchestrator picks up. orchestrator picks up.
@@ -125,7 +147,7 @@ class CronScheduler:
self.scheduler.reschedule_job(job_id, trigger=trigger) self.scheduler.reschedule_job(job_id, trigger=trigger)
else: else:
self.scheduler.add_job( self.scheduler.add_job(
self.trigger, dispatch_persisted_cron,
trigger=trigger, trigger=trigger,
args=[item.schedule_id], args=[item.schedule_id],
id=job_id, id=job_id,
+2 -2
View File
@@ -372,14 +372,14 @@ class NodeExecutor:
return log_id, result_id, upload_error return log_id, result_id, upload_error
@staticmethod @staticmethod
def _start_inbox( async def _start_inbox(
session, session,
*, *,
consumer_name: str, consumer_name: str,
event_id: str, event_id: str,
message_id: str, message_id: str,
) -> tuple[ConsumerInbox, bool]: ) -> tuple[ConsumerInbox, bool]:
item = session.get( item = await session.get(
ConsumerInbox, ConsumerInbox,
(consumer_name, event_id), (consumer_name, event_id),
with_for_update=True, with_for_update=True,