schedule/src/schedule/service.py

@@ -21,6 +21,7 @@ service shares the same MySQL via the Docker network).
 from __future__ import annotations
 
 from datetime import UTC, datetime
+from pathlib import Path
 from typing import Any
 
 import httpx
@@ -188,16 +189,31 @@ class SchedulerService:
 
 
 def build_object_store() -> Any:
-    """Construct an AsyncStorageBackend pointed at S3.
+    """Construct the AsyncStorageBackend the worker reads version artifacts from.
 
-    Reads ``s3_endpoint`` / ``s3_access_key`` / ``s3_secret_key``
-    from :data:`common.config.settings`.
+    Respects ``settings.storage_backend``: in ``local`` mode points at the
+    local ``${local_storage_base_dir}/version`` directory (the same place
+    ``backend.storage_api`` writes to), in ``s3`` mode points at
+    ``settings.s3_version_bucket``. The worker only reads version artifacts
+    via this store (run logs / results still go through the backend's
+    HTTP storage API), so pointing at the version bucket/directory is the
+    correct resolution regardless of the artifact's workspace.
     """
+    if settings.storage_backend == "local":
+        return create_storage(
+            {
+                "type": "local",
+                "mode": "async",
+                "base_dir": str(
+                    Path(settings.local_storage_base_dir) / "version"
+                ),
+            }
+        )
     return create_storage(
         {
             "type": "s3",
             "mode": "async",
-            "bucket": settings.s3_workspace_bucket,
+            "bucket": settings.s3_version_bucket,
             "endpoint_url": settings.s3_endpoint,
             "aws_access_key_id": settings.s3_access_key,
             "aws_secret_access_key": settings.s3_secret_key,

schedule/src/schedule/worker.py

@@ -23,6 +23,7 @@ from typing import Any
 from loguru import logger
 from sqlalchemy import select
 
+from common.config import settings
 from common.db import session_scope
 from common.db.models import (
     ConsumerInbox,
@@ -295,8 +296,10 @@ class NodeExecutor:
             node_run, run, version, storage, workspace, schedule = row
             if storage.object_status != "available":
                 raise ValueError("stable version artifact is not available")
-            if storage.storage_backend != "s3":
-                raise ValueError("stable version artifact is not stored in S3")
+            if storage.storage_backend != settings.storage_backend:
+                raise ValueError(
+                    "稳定版本产物的存储后端与当前运行后端不一致"
+                )
             if not storage.bucket_name or not storage.object_key:
                 raise ValueError("stable version artifact location is incomplete")
             user_id = run.triggered_by or schedule.created_by