diff --git a/schedule/src/schedule/service.py b/schedule/src/schedule/service.py index decf2b8..c14a73d 100644 --- a/schedule/src/schedule/service.py +++ b/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, diff --git a/schedule/src/schedule/worker.py b/schedule/src/schedule/worker.py index df61ad2..b0cf1bb 100644 --- a/schedule/src/schedule/worker.py +++ b/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