feat: 新增 MinIO 对象存储与算力节点缓存预下载

- 后端新增 storage 模块（minio_store），支持 MinIO 预签名 URL 上传与对象管理 - config 新增 MinIO 及存储等待相关配置项 - 算力节点新增 /compute/cache/prepare 缓存预下载接口（带校验和原子落盘） - 算力节点健康接口增加存储可用性探针 - SQL 迁移补充资源存储相关表结构 - Docker 新增 minio 服务与后端 minio 配置 - 补充 minio-compute-cache-plan 设计文档 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-11 16:25:18 +08:00
parent a12f80492d
commit b5d2cd7935
20 changed files with 830 additions and 11 deletions
--- a/backend/app/api/v1/endpoints/platform.py
+++ b/backend/app/api/v1/endpoints/platform.py
@@ -1,6 +1,8 @@
 from __future__ import annotations

 import json
+import asyncio
+import hashlib
 import uuid
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -16,6 +18,7 @@ from app.core.config import get_settings
 from app.db.platform_store import get_platform_store
 from app.modules.compute_gateway.client import ComputeNodeClient
 from app.modules.compute_gateway.sync import fetch_eval_result_content, poll_compute_jobs_once
+from app.modules.storage.minio_store import ObjectStorageError, get_object_storage

 router = APIRouter()

@@ -33,6 +36,22 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None:
    return None


+async def _wait_for_object_storage() -> None:
+    """Wait for MinIO before starting a resource task."""
+    settings = get_settings()
+    deadline = datetime.now(timezone.utc).timestamp() + max(0, settings.storage_wait_seconds)
+    last_error = "MinIO unavailable"
+    while True:
+        try:
+            get_object_storage().ensure_bucket()
+            return
+        except Exception as exc:  # noqa: BLE001 - retry until the configured deadline
+            last_error = str(exc)
+        if datetime.now(timezone.utc).timestamp() >= deadline:
+            raise RuntimeError(f"MinIO unavailable after {settings.storage_wait_seconds}s: {last_error}")
+        await asyncio.sleep(max(1, settings.storage_check_interval_seconds))
+
+
 def _select_eval_node(store: Any, preferred_node_id: str | None = None) -> dict[str, Any] | None:
    """Select the compute node for an eval job.

@@ -330,6 +349,11 @@ async def _fine_tune_preflight_with_job_payload(
            )
        except Exception as exc:  # noqa: BLE001 - return as preflight error for page visibility
            sync_errors.append(str(exc))
+    if get_settings().minio_enabled and get_settings().compute_mode != "simulator":
+        try:
+            await _wait_for_object_storage()
+        except Exception as exc:  # noqa: BLE001 - preflight exposes node storage failure
+            sync_errors.append(f"shared storage health check failed: {exc}")
    if get_settings().compute_mode == "simulator":
        preview = {
            "valid": True,
@@ -877,6 +901,11 @@ async def merge_model(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
        raise fail(400, "adapter_path is required")
    requested_node_id = payload.get("requested_node_id") or payload.get("compute_node_id") or (trained_model and trained_model.get("compute_node_id"))
    node = store.schedule_node({**payload, "requested_node_id": requested_node_id, "gpus": payload.get("gpus") or []})
+    if get_settings().minio_enabled and get_settings().compute_mode != "simulator":
+        try:
+            await _wait_for_object_storage()
+        except RuntimeError as exc:
+            raise fail(503, str(exc))
    health = node.get("health_detail") or {}
    output_root = str(health.get("output_root") or f"{node['data_root'].rstrip('/')}/outputs")
    output_name = str(payload.get("output_model_name") or payload.get("merged_model_name") or f"{trained_model_id or 'model'}-merged")
@@ -981,7 +1010,7 @@ async def _sync_dataset_file_to_compute_nodes(
    content: bytes,
 ) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
-    if get_settings().compute_mode == "simulator":
+    if get_settings().compute_mode == "simulator" or get_settings().minio_enabled:
        return results
    target_name = Path(filename or f"{file_id}.jsonl").name
    target_relative_path = f"datasets/{dataset_id}/{target_name}"
@@ -1029,6 +1058,27 @@ async def _sync_training_dataset_to_compute_node(
    node: dict[str, Any],
    dataset_id: str,
 ) -> list[dict[str, Any]]:
+    if get_settings().minio_enabled:
+        files = store.training_dataset_files(dataset_id)
+        objects = store.storage_objects_for_resource("dataset", dataset_id)
+        object_by_name = {Path(str(item.get("file_name") or item.get("object_key") or "")).name: item for item in objects}
+        results: list[dict[str, Any]] = []
+        client = ComputeNodeClient(node["api_base_url"])
+        for item in files:
+            target_name = Path(str(item.get("name") or f"{item['id']}.jsonl")).name
+            obj = object_by_name.get(target_name)
+            if not obj:
+                raise RuntimeError(f"dataset file is not available in MinIO: {target_name}")
+            url = get_object_storage().presigned_get(obj["object_key"])
+            result = await client.prepare_cache({
+                "resource_id": dataset_id,
+                "version_id": obj["version_id"],
+                "download_url": url,
+                "checksum_sha256": obj.get("checksum_sha256") or "",
+                "relative_path": f"datasets/{dataset_id}/{target_name}",
+            })
+            results.append({**result, "file_id": item.get("id"), "name": target_name, "node_id": node["id"]})
+        return results
    if not dataset_id:
        raise RuntimeError("train_dataset_id is required")
    files = store.training_dataset_files(dataset_id)
@@ -1092,6 +1142,17 @@ async def upload_dataset_files(
            created_file = store.add_dataset_file(conn, dataset_id, file.filename or "upload.jsonl", content)
            created.append(created_file)
            pending_sync.append((created_file["id"], created_file["name"], raw))
+            if get_settings().minio_enabled:
+                object_key = f"datasets/{dataset_id}/versions/{created_file.get('active_version_id') or created_file['id']}/{Path(created_file['name']).name}"
+                uploaded = get_object_storage().put_bytes(object_key, raw, file.content_type or "application/octet-stream")
+                get_platform_store().create_storage_object({
+                    "resource_type": "dataset", "resource_id": dataset_id,
+                    "version_id": created_file.get("active_version_id") or created_file["id"],
+                    "bucket": uploaded["bucket"], "object_key": object_key,
+                    "file_name": created_file["name"], "content_type": file.content_type,
+                    "byte_size": len(raw), "checksum_sha256": hashlib.sha256(raw).hexdigest(),
+                    "status": "available",
+                })
    if sync_to_compute:
        for file_id, file_name, raw in pending_sync:
            compute_sync.extend(
@@ -1865,6 +1926,8 @@ async def model_compare_load(task_id: str) -> dict[str, Any]:
            errors = []
            for node in _candidate_online_nodes(store, preferred_node_id):
                try:
+                    if get_settings().minio_enabled:
+                        await _wait_for_object_storage()
                    client = ComputeNodeClient(node["api_base_url"])
                    await client.inference_load(load_payload)
                    store.mark_inference_loaded(node["id"])
@@ -2035,6 +2098,73 @@ async def compute_nodes() -> dict[str, Any]:
    return ok(get_platform_store().compute_nodes())


+@router.post("/storage/objects/presign")
+async def presign_storage_object(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
+    """Create a short-lived MinIO upload/download URL for a platform resource."""
+    if not get_settings().minio_enabled:
+        raise fail(503, "MinIO object storage is disabled")
+    resource_type = str(payload.get("resource_type") or "")
+    resource_id = str(payload.get("resource_id") or "")
+    version_id = str(payload.get("version_id") or uuid.uuid4().hex)
+    object_key = str(payload.get("object_key") or f"{resource_type}/{resource_id}/versions/{version_id}/resource")
+    if not resource_type or not resource_id:
+        raise fail(400, "resource_type and resource_id are required")
+    if payload.get("method", "put").lower() == "get" and not has_resource_access(resource_type, resource_id, current_user, "read"):
+        raise fail(403, "no permission to read this resource")
+    try:
+        storage = get_object_storage()
+        url = storage.presigned_get(object_key) if payload.get("method", "put").lower() == "get" else storage.presigned_put(object_key)
+        record = get_platform_store().create_storage_object({
+            "resource_type": resource_type, "resource_id": resource_id, "version_id": version_id,
+            "bucket": storage.bucket, "object_key": object_key, "file_name": payload.get("file_name"),
+            "content_type": payload.get("content_type"), "created_by": current_user.get("id"),
+        })
+        return ok({"url": url, "method": payload.get("method", "put").lower(), "expires_seconds": 3600, "object": record})
+    except ObjectStorageError as exc:
+        raise fail(503, str(exc))
+
+
+@router.get("/storage/resources/{resource_type}/{resource_id}")
+async def storage_resource_objects(resource_type: str, resource_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
+    if not has_resource_access(resource_type, resource_id, current_user, "read"):
+        raise fail(403, "no permission to read this resource")
+    return ok(get_platform_store().storage_objects_for_resource(resource_type, resource_id))
+
+
+@router.post("/storage/resources/{resource_type}/{resource_id}/prepare/{node_id}")
+async def prepare_storage_resource(
+    resource_type: str,
+    resource_id: str,
+    node_id: str,
+    current_user: dict = Depends(get_current_user),
+) -> dict[str, Any]:
+    if not has_resource_access(resource_type, resource_id, current_user, "execute"):
+        raise fail(403, "no permission to execute this resource")
+    store = get_platform_store()
+    node = next((item for item in store.compute_nodes() if item["id"] == node_id), None)
+    if not node:
+        raise fail(404, "compute node not found")
+    if not get_settings().minio_enabled:
+        raise fail(503, "MinIO object storage is disabled")
+    objects = store.storage_objects_for_resource(resource_type, resource_id)
+    if not objects:
+        raise fail(404, "resource has no MinIO objects")
+    client = ComputeNodeClient(node["api_base_url"])
+    prepared = []
+    for obj in objects:
+        url = get_object_storage().presigned_get(obj["object_key"])
+        filename = Path(str(obj.get("file_name") or obj["object_key"])).name
+        result = await client.prepare_cache({
+            "resource_id": resource_id,
+            "version_id": obj["version_id"],
+            "download_url": url,
+            "checksum_sha256": obj.get("checksum_sha256") or "",
+            "relative_path": f"{resource_type}s/{resource_id}/{filename}",
+        })
+        prepared.append({**result, "storage_object_id": obj["id"], "node_id": node_id})
+    return ok({"resource_type": resource_type, "resource_id": resource_id, "node_id": node_id, "status": "ready", "items": prepared})
+
+
@router.get("/compute/nodes/{node_id}")
 async def compute_node_detail(node_id: str) -> dict[str, Any]:
    node = next((item for item in get_platform_store().compute_nodes() if item["id"] == node_id), None)
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -60,6 +60,14 @@ class Settings:
    compute_status_sync_mode: str = os.getenv("COMPUTE_STATUS_SYNC_MODE", "polling")
    compute_poll_interval_seconds: int = _int_env("COMPUTE_POLL_INTERVAL_SECONDS", 3)
    compute_request_timeout_seconds: int = _int_env("COMPUTE_REQUEST_TIMEOUT_SECONDS", 5)
+    minio_enabled: bool = _bool_env("MINIO_ENABLED", False)
+    minio_endpoint: str = os.getenv("MINIO_ENDPOINT", "http://minio:9000")
+    minio_access_key: str = os.getenv("MINIO_ACCESS_KEY", "minioadmin")
+    minio_secret_key: str = os.getenv("MINIO_SECRET_KEY", "minioadmin")
+    minio_bucket: str = os.getenv("MINIO_BUCKET", "yg-ft-resources")
+    minio_secure: bool = _bool_env("MINIO_SECURE", False)
+    storage_wait_seconds: int = _int_env("STORAGE_WAIT_SECONDS", 300)
+    storage_check_interval_seconds: int = _int_env("STORAGE_CHECK_INTERVAL_SECONDS", 10)
    compute_service_token: str = os.getenv("COMPUTE_SERVICE_TOKEN", "")
    log_level: str = os.getenv("LOG_LEVEL", "INFO")
    log_dir: str = os.getenv("LOG_DIR", "./logs")
--- a/backend/app/db/platform_store.py
+++ b/backend/app/db/platform_store.py
@@ -505,6 +505,20 @@ class PlatformStore:
            )
            self._ensure_columns(conn, "gpus", {"last_seen_at": "TEXT"})
            self._ensure_columns(conn, "fine_tune_tasks", {"compute_job_id": "TEXT"})
+            self._ensure_columns(
+                conn,
+                "sessions",
+                {
+                    "username": "TEXT",
+                    "login_at": "TEXT",
+                    "logout_at": "TEXT",
+                    "duration_seconds": "INTEGER",
+                    "issued_at": "TEXT",
+                    "expires_at": "TEXT",
+                    "ip": "TEXT",
+                    "create_time": "TEXT",
+                },
+            )
            self._ensure_columns(
                conn,
                "trained_models",
@@ -525,7 +539,12 @@ class PlatformStore:
                },
            )
            schema_dir = Path(__file__).with_name("sql")
-            for extra in ("002_governance.sql", "003_tenant_quota.sql", "004_permissions.sql"):
+            for extra in (
+                "002_governance.sql",
+                "003_model_path_governance.sql",
+                "003_tenant_quota.sql",
+                "004_permissions.sql",
+            ):
                extra_path = schema_dir / extra
                if extra_path.exists():
                    conn.executescript(extra_path.read_text(encoding="utf-8"))
@@ -2863,6 +2882,54 @@ class PlatformStore:
                    result.append(dict(row))
            return result

+    def create_storage_object(self, payload: dict[str, Any]) -> dict[str, Any]:
+        object_id = str(payload.get("id") or new_id("object"))
+        with self.connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO storage_objects
+                (id, resource_type, resource_id, version_id, bucket, object_key, file_name,
+                 content_type, checksum_sha256, byte_size, status, created_by, create_time)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT (resource_type, resource_id, version_id, object_key)
+                DO UPDATE SET file_name=EXCLUDED.file_name, content_type=EXCLUDED.content_type,
+                    checksum_sha256=EXCLUDED.checksum_sha256, byte_size=EXCLUDED.byte_size,
+                    status=EXCLUDED.status, created_by=EXCLUDED.created_by
+                """,
+                (
+                    object_id, payload["resource_type"], payload["resource_id"], payload["version_id"],
+                    payload["bucket"], payload["object_key"], payload.get("file_name"), payload.get("content_type"),
+                    payload.get("checksum_sha256"), int(payload.get("byte_size") or 0), payload.get("status", "pending"),
+                    payload.get("created_by"), payload.get("create_time") or utcnow(),
+                ),
+            )
+            row = conn.execute(
+                "SELECT * FROM storage_objects WHERE resource_type=? AND resource_id=? AND version_id=? AND object_key=?",
+                (payload["resource_type"], payload["resource_id"], payload["version_id"], payload["object_key"]),
+            ).fetchone()
+            return dict(row)
+
+    def storage_objects_for_resource(self, resource_type: str, resource_id: str) -> list[dict[str, Any]]:
+        with self.connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM storage_objects WHERE resource_type=? AND resource_id=? AND status='available' ORDER BY version_id, object_key",
+                (resource_type, resource_id),
+            ).fetchall()
+            return [dict(row) for row in rows]
+
+    def update_storage_object(self, object_id: str, payload: dict[str, Any]) -> dict[str, Any]:
+        allowed = {"status", "checksum_sha256", "byte_size", "content_type"}
+        fields = {key: value for key, value in payload.items() if key in allowed}
+        if fields:
+            assignments = ", ".join(f"{key}=?" for key in fields)
+            with self.connect() as conn:
+                conn.execute(f"UPDATE storage_objects SET {assignments} WHERE id=?", (*fields.values(), object_id))
+        with self.connect() as conn:
+            row = conn.execute("SELECT * FROM storage_objects WHERE id=?", (object_id,)).fetchone()
+            if not row:
+                raise KeyError(object_id)
+            return dict(row)
+
    def update_resource_replica_sync_result(
        self,
        replica_id: str,
--- a/backend/app/db/sql/000_full_init.sql
+++ b/backend/app/db/sql/000_full_init.sql
@@ -276,6 +276,39 @@ CREATE TABLE IF NOT EXISTS resource_sync_jobs (
    completed_at TEXT
 );

+CREATE TABLE IF NOT EXISTS storage_objects (
+    id TEXT PRIMARY KEY,
+    resource_type TEXT NOT NULL,
+    resource_id TEXT NOT NULL,
+    version_id TEXT NOT NULL,
+    bucket TEXT NOT NULL,
+    object_key TEXT NOT NULL,
+    file_name TEXT,
+    content_type TEXT,
+    checksum_sha256 TEXT,
+    byte_size BIGINT NOT NULL DEFAULT 0,
+    status TEXT NOT NULL DEFAULT 'pending',
+    created_by TEXT,
+    create_time TEXT NOT NULL,
+    UNIQUE (resource_type, resource_id, version_id, object_key)
+);
+
+CREATE TABLE IF NOT EXISTS storage_cache_jobs (
+    id TEXT PRIMARY KEY,
+    storage_object_id TEXT NOT NULL REFERENCES storage_objects(id) ON DELETE CASCADE,
+    node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE,
+    direction TEXT NOT NULL,
+    status TEXT NOT NULL,
+    progress INTEGER NOT NULL DEFAULT 0,
+    local_path TEXT,
+    error TEXT,
+    create_time TEXT NOT NULL,
+    completed_at TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_storage_objects_resource ON storage_objects(resource_type, resource_id, version_id);
+CREATE INDEX IF NOT EXISTS idx_storage_cache_jobs_node_status ON storage_cache_jobs(node_id, status);
+
 CREATE TABLE IF NOT EXISTS eval_tasks (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
@@ -361,9 +394,14 @@ CREATE TABLE IF NOT EXISTS roles (
 CREATE TABLE IF NOT EXISTS sessions (
    id TEXT PRIMARY KEY,
    user_id TEXT NOT NULL,
-    issued_at TEXT NOT NULL,
-    expires_at TEXT NOT NULL,
-    ip TEXT
+    username TEXT,
+    login_at TEXT,
+    logout_at TEXT,
+    duration_seconds INTEGER,
+    issued_at TEXT,
+    expires_at TEXT,
+    ip TEXT,
+    create_time TEXT
 );

 CREATE TABLE IF NOT EXISTS acls (
@@ -376,6 +414,25 @@ CREATE TABLE IF NOT EXISTS acls (
    create_time TEXT
 );

+-- ---- 权限扩展（来源：004_permissions.sql） ----
+
+CREATE TABLE IF NOT EXISTS gpu_assignments (
+    id TEXT PRIMARY KEY,
+    node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE,
+    gpu_index INTEGER NOT NULL,
+    user_id TEXT NOT NULL,
+    assigned_by TEXT,
+    assigned_at TEXT NOT NULL,
+    UNIQUE (node_id, gpu_index, user_id)
+);
+CREATE INDEX IF NOT EXISTS idx_gpu_assignments_user ON gpu_assignments(user_id);
+CREATE INDEX IF NOT EXISTS idx_gpu_assignments_gpu ON gpu_assignments(node_id, gpu_index);
+
+ALTER TABLE datasets ADD COLUMN IF NOT EXISTS created_by TEXT;
+ALTER TABLE models ADD COLUMN IF NOT EXISTS created_by TEXT;
+ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS created_by TEXT;
+ALTER TABLE eval_tasks ADD COLUMN IF NOT EXISTS created_by TEXT;
+
 -- ============================================================================
 -- 二、治理表（来源：002_governance.sql）
 -- ============================================================================
@@ -462,6 +519,8 @@ ALTER TABLE tenants ADD COLUMN IF NOT EXISTS storage_quota TEXT;

 ALTER TABLE models ADD COLUMN IF NOT EXISTS can_train INTEGER NOT NULL DEFAULT 0;
 ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS artifact_dir TEXT;
+ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_id TEXT;
+ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_name TEXT;

 -- 按规则推定已有模型的 can_train（新库为空表，此语句为 no-op）
 UPDATE models
--- a/backend/app/db/sql/001_platform_runtime.sql
+++ b/backend/app/db/sql/001_platform_runtime.sql
@@ -234,6 +234,39 @@ CREATE TABLE IF NOT EXISTS resource_sync_jobs (
    completed_at TEXT
 );

+CREATE TABLE IF NOT EXISTS storage_objects (
+    id TEXT PRIMARY KEY,
+    resource_type TEXT NOT NULL,
+    resource_id TEXT NOT NULL,
+    version_id TEXT NOT NULL,
+    bucket TEXT NOT NULL,
+    object_key TEXT NOT NULL,
+    file_name TEXT,
+    content_type TEXT,
+    checksum_sha256 TEXT,
+    byte_size BIGINT NOT NULL DEFAULT 0,
+    status TEXT NOT NULL DEFAULT 'pending',
+    created_by TEXT,
+    create_time TEXT NOT NULL,
+    UNIQUE (resource_type, resource_id, version_id, object_key)
+);
+
+CREATE TABLE IF NOT EXISTS storage_cache_jobs (
+    id TEXT PRIMARY KEY,
+    storage_object_id TEXT NOT NULL REFERENCES storage_objects(id) ON DELETE CASCADE,
+    node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE,
+    direction TEXT NOT NULL,
+    status TEXT NOT NULL,
+    progress INTEGER NOT NULL DEFAULT 0,
+    local_path TEXT,
+    error TEXT,
+    create_time TEXT NOT NULL,
+    completed_at TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_storage_objects_resource ON storage_objects(resource_type, resource_id, version_id);
+CREATE INDEX IF NOT EXISTS idx_storage_cache_jobs_node_status ON storage_cache_jobs(node_id, status);
+
 CREATE TABLE IF NOT EXISTS eval_tasks (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
@@ -319,9 +352,14 @@ CREATE TABLE IF NOT EXISTS roles (
 CREATE TABLE IF NOT EXISTS sessions (
    id TEXT PRIMARY KEY,
    user_id TEXT NOT NULL,
-    issued_at TEXT NOT NULL,
-    expires_at TEXT NOT NULL,
-    ip TEXT
+    username TEXT,
+    login_at TEXT,
+    logout_at TEXT,
+    duration_seconds INTEGER,
+    issued_at TEXT,
+    expires_at TEXT,
+    ip TEXT,
+    create_time TEXT
 );

 CREATE TABLE IF NOT EXISTS acls (
--- a/backend/app/db/sql/003_model_path_governance.sql
+++ b/backend/app/db/sql/003_model_path_governance.sql
@@ -16,3 +16,5 @@ END;

 -- 3. 给 trained_models 增加 artifact_dir（训练产物目录扫描结果目录）
 ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS artifact_dir TEXT;
+ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_id TEXT;
+ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS compute_node_name TEXT;
--- a/backend/app/modules/compute_gateway/client.py
+++ b/backend/app/modules/compute_gateway/client.py
@@ -192,6 +192,27 @@ class ComputeNodeClient:
            response.raise_for_status()
            return _unwrap_dict(response.json())

+    async def prepare_cache(self, payload: dict[str, Any]) -> dict[str, Any]:
+        async with httpx.AsyncClient(timeout=httpx.Timeout(900, connect=30), headers=self.headers()) as client:
+            response = await client.post(
+                _join_url(self.api_base_url, f"{self.route_prefix}/compute/cache/prepare"),
+                json=payload,
+            )
+            response.raise_for_status()
+            return _unwrap_dict(response.json())
+
+    async def cache_status(self, resource_id: str, version_id: str | None = None) -> dict[str, Any]:
+        params = {"resource_id": resource_id}
+        if version_id:
+            params["version_id"] = version_id
+        async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client:
+            response = await client.get(
+                _join_url(self.api_base_url, f"{self.route_prefix}/compute/cache/status"),
+                params=params,
+            )
+            response.raise_for_status()
+            return _unwrap_dict(response.json())
+
    async def _request(
        self,
        method: str,
--- a/backend/app/modules/storage/init.py
+++ b/backend/app/modules/storage/init.py
@@ -0,0 +1 @@
+"""Central object storage integration."""
--- a/backend/app/modules/storage/minio_store.py
+++ b/backend/app/modules/storage/minio_store.py
@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from datetime import timedelta
+from functools import lru_cache
+from io import BytesIO
+from typing import Any
+
+from minio import Minio
+from minio.error import S3Error
+
+from app.core.config import get_settings
+
+
+class ObjectStorageError(RuntimeError):
+    pass
+
+
+class MinioObjectStorage:
+    def __init__(self) -> None:
+        settings = get_settings()
+        endpoint = settings.minio_endpoint.replace("http://", "").replace("https://", "").rstrip("/")
+        self.client = Minio(endpoint, access_key=settings.minio_access_key, secret_key=settings.minio_secret_key, secure=settings.minio_secure)
+        self.bucket = settings.minio_bucket
+
+    def _ensure_enabled(self) -> None:
+        if not get_settings().minio_enabled:
+            raise ObjectStorageError("MinIO object storage is disabled")
+
+    def ensure_bucket(self) -> None:
+        self._ensure_enabled()
+        try:
+            if not self.client.bucket_exists(self.bucket):
+                self.client.make_bucket(self.bucket)
+        except S3Error as exc:
+            raise ObjectStorageError(str(exc)) from exc
+
+    def presigned_put(self, object_key: str, expires_seconds: int = 3600) -> str:
+        self._ensure_enabled()
+        self.ensure_bucket()
+        return self.client.presigned_put_object(self.bucket, object_key, expires=timedelta(seconds=expires_seconds))
+
+    def presigned_get(self, object_key: str, expires_seconds: int = 3600) -> str:
+        self._ensure_enabled()
+        self.ensure_bucket()
+        return self.client.presigned_get_object(self.bucket, object_key, expires=timedelta(seconds=expires_seconds))
+
+    def stat(self, object_key: str) -> dict[str, Any]:
+        self._ensure_enabled()
+        self.ensure_bucket()
+        try:
+            result = self.client.stat_object(self.bucket, object_key)
+            return {"object_key": object_key, "byte_size": result.size, "etag": result.etag, "last_modified": result.last_modified.isoformat() if result.last_modified else None}
+        except S3Error as exc:
+            raise ObjectStorageError(str(exc)) from exc
+
+    def put_bytes(self, object_key: str, content: bytes, content_type: str = "application/octet-stream") -> dict[str, Any]:
+        self._ensure_enabled()
+        self.ensure_bucket()
+        try:
+            result = self.client.put_object(self.bucket, object_key, BytesIO(content), len(content), content_type=content_type)
+            return {"bucket": self.bucket, "object_key": object_key, "etag": result.etag, "byte_size": len(content)}
+        except S3Error as exc:
+            raise ObjectStorageError(str(exc)) from exc
+
+
+@lru_cache
+def get_object_storage() -> MinioObjectStorage:
+    return MinioObjectStorage()