fix: initialize object storage before API readiness

This commit is contained in:
2026-09-20 23:22:05 +08:00
parent 7b56bce900
commit cac183b2ea
3 changed files with 175 additions and 5 deletions
@@ -68,6 +68,9 @@ def create_app(
secret_access_key=resolved_settings.object_storage_secret_access_key,
region_name=resolved_settings.object_storage_region_name,
)
# A fresh deployment must establish its storage namespace before the API
# can report healthy. This also makes empty object stores backup-ready.
object_storage.ensure_ready()
anti_abuse_guard = AntiAbuseGuard(
mode=resolved_settings.anti_abuse_mode,
service_name=resolved_settings.service_name,
@@ -16,6 +16,8 @@ class StoredObject:
class ObjectStorage(Protocol):
def ensure_ready(self) -> None: ...
def put_bytes(self, *, filename: str, content: bytes) -> StoredObject: ...
def get_bytes(self, *, bucket: str, object_key: str) -> bytes: ...
@@ -28,7 +30,11 @@ class LocalObjectStorage:
self.root_dir = Path(root_dir)
self.bucket = bucket
def ensure_ready(self) -> None:
(self.root_dir / self.bucket).mkdir(parents=True, exist_ok=True)
def put_bytes(self, *, filename: str, content: bytes) -> StoredObject:
self.ensure_ready()
suffix = Path(filename).suffix
object_key = f"knowledge-ingestion/{uuid4().hex}{suffix}"
target = self.root_dir / self.bucket / object_key
@@ -81,7 +87,7 @@ class S3CompatibleObjectStorage:
)
def put_bytes(self, *, filename: str, content: bytes) -> StoredObject:
self._ensure_bucket()
self.ensure_ready()
suffix = Path(filename).suffix
object_key = f"knowledge-ingestion/{uuid4().hex}{suffix}"
content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
@@ -105,7 +111,7 @@ class S3CompatibleObjectStorage:
def delete_object(self, *, bucket: str, object_key: str) -> None:
self.client.delete_object(Bucket=bucket, Key=object_key)
def _ensure_bucket(self) -> None:
def ensure_ready(self) -> None:
if self._bucket_checked:
return
try:
@@ -114,7 +120,15 @@ class S3CompatibleObjectStorage:
code = _s3_error_code(error)
if code not in {"404", "NoSuchBucket", "NotFound"}:
raise
self.client.create_bucket(Bucket=self.bucket)
try:
self.client.create_bucket(Bucket=self.bucket)
except Exception as create_error: # pragma: no cover - covered through fakes
# Multiple API workers can observe a missing bucket at the same time.
# Only the provider's explicit ownership response followed by a
# successful read-back is safe to accept.
if _s3_error_code(create_error) != "BucketAlreadyOwnedByYou":
raise
self.client.head_bucket(Bucket=self.bucket)
self._bucket_checked = True
@@ -3,6 +3,8 @@ from types import ModuleType, SimpleNamespace
import pytest
from yuqei_ai_platform_api.object_storage import LocalObjectStorage
class FakeS3Error(Exception):
def __init__(self, code: str) -> None:
@@ -34,9 +36,27 @@ class FakeS3Client:
self.objects.pop((Bucket, Key), None)
def test_s3_storage_creates_missing_bucket_before_first_upload(monkeypatch: pytest.MonkeyPatch) -> None:
fake_client = FakeS3Client()
class HeadFailingS3Client(FakeS3Client):
def head_bucket(self, *, Bucket: str) -> None:
raise FakeS3Error("AccessDenied")
class RacingS3Client(FakeS3Client):
def head_bucket(self, *, Bucket: str) -> None:
self.head_calls += 1
if self.head_calls == 1:
raise FakeS3Error("NoSuchBucket")
def create_bucket(self, *, Bucket: str) -> None:
raise FakeS3Error("BucketAlreadyOwnedByYou")
class CreateFailingS3Client(FakeS3Client):
def create_bucket(self, *, Bucket: str) -> None:
raise FakeS3Error("AccessDenied")
def _install_fake_boto3(monkeypatch: pytest.MonkeyPatch, fake_client: FakeS3Client) -> None:
fake_boto3 = ModuleType("boto3")
fake_boto3.client = lambda *args, **kwargs: fake_client # type: ignore[attr-defined]
fake_botocore = ModuleType("botocore")
@@ -47,6 +67,40 @@ def test_s3_storage_creates_missing_bucket_before_first_upload(monkeypatch: pyte
monkeypatch.setitem(sys.modules, "botocore", fake_botocore)
monkeypatch.setitem(sys.modules, "botocore.client", fake_botocore_client)
def test_local_storage_ensure_ready_creates_empty_bucket(tmp_path) -> None:
storage = LocalObjectStorage(tmp_path, bucket="ai-platform-uploads")
storage.ensure_ready()
assert (tmp_path / "ai-platform-uploads").is_dir()
def test_create_app_initializes_empty_object_storage_bucket(tmp_path) -> None:
from yuqei_ai_platform_api.config import AiPlatformSettings
from yuqei_ai_platform_api.main import create_app
from yuqei_ai_platform_api.repository import InMemoryAiPlatformRepository
object_root = tmp_path / "objects"
create_app(
AiPlatformSettings(
_env_file=None,
environment="test",
repository_backend="memory",
object_storage_backend="local",
object_storage_local_dir=str(object_root),
object_storage_bucket="fresh-install-bucket",
),
repository=InMemoryAiPlatformRepository(),
)
assert (object_root / "fresh-install-bucket").is_dir()
def test_s3_storage_ensure_ready_creates_missing_bucket_once(monkeypatch: pytest.MonkeyPatch) -> None:
fake_client = FakeS3Client()
_install_fake_boto3(monkeypatch, fake_client)
from yuqei_ai_platform_api.object_storage import S3CompatibleObjectStorage
storage = S3CompatibleObjectStorage(
@@ -56,6 +110,8 @@ def test_s3_storage_creates_missing_bucket_before_first_upload(monkeypatch: pyte
bucket="ai-platform-uploads",
)
storage.ensure_ready()
storage.ensure_ready()
stored = storage.put_bytes(filename="law.txt", content=b"contract law")
second = storage.put_bytes(filename="policy.txt", content=b"policy")
@@ -63,3 +119,100 @@ def test_s3_storage_creates_missing_bucket_before_first_upload(monkeypatch: pyte
assert fake_client.head_calls == 1
assert storage.get_bytes(bucket=stored.bucket, object_key=stored.object_key) == b"contract law"
assert storage.get_bytes(bucket=second.bucket, object_key=second.object_key) == b"policy"
def test_s3_storage_ensure_ready_reuses_existing_bucket(monkeypatch: pytest.MonkeyPatch) -> None:
fake_client = FakeS3Client()
fake_client.created_buckets.append("ai-platform-uploads")
_install_fake_boto3(monkeypatch, fake_client)
from yuqei_ai_platform_api.object_storage import S3CompatibleObjectStorage
storage = S3CompatibleObjectStorage(
endpoint_url="http://minio:9000",
access_key_id="access",
secret_access_key="secret",
bucket="ai-platform-uploads",
)
storage.ensure_ready()
assert fake_client.head_calls == 1
assert fake_client.created_buckets == ["ai-platform-uploads"]
def test_s3_storage_ensure_ready_rejects_unexpected_head_error(monkeypatch: pytest.MonkeyPatch) -> None:
fake_client = HeadFailingS3Client()
_install_fake_boto3(monkeypatch, fake_client)
from yuqei_ai_platform_api.object_storage import S3CompatibleObjectStorage
storage = S3CompatibleObjectStorage(
endpoint_url="http://minio:9000",
access_key_id="access",
secret_access_key="secret",
bucket="ai-platform-uploads",
)
with pytest.raises(FakeS3Error, match="AccessDenied"):
storage.ensure_ready()
def test_s3_storage_ensure_ready_accepts_owned_bucket_creation_race(monkeypatch: pytest.MonkeyPatch) -> None:
fake_client = RacingS3Client()
_install_fake_boto3(monkeypatch, fake_client)
from yuqei_ai_platform_api.object_storage import S3CompatibleObjectStorage
storage = S3CompatibleObjectStorage(
endpoint_url="http://minio:9000",
access_key_id="access",
secret_access_key="secret",
bucket="ai-platform-uploads",
)
storage.ensure_ready()
storage.ensure_ready()
assert fake_client.head_calls == 2
def test_s3_storage_ensure_ready_retries_after_create_failure(monkeypatch: pytest.MonkeyPatch) -> None:
fake_client = CreateFailingS3Client()
_install_fake_boto3(monkeypatch, fake_client)
from yuqei_ai_platform_api.object_storage import S3CompatibleObjectStorage
storage = S3CompatibleObjectStorage(
endpoint_url="http://minio:9000",
access_key_id="access",
secret_access_key="secret",
bucket="ai-platform-uploads",
)
with pytest.raises(FakeS3Error, match="AccessDenied"):
storage.ensure_ready()
with pytest.raises(FakeS3Error, match="AccessDenied"):
storage.ensure_ready()
assert fake_client.head_calls == 2
def test_create_app_fails_closed_when_object_storage_is_not_ready(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from yuqei_ai_platform_api import main
from yuqei_ai_platform_api.config import AiPlatformSettings
from yuqei_ai_platform_api.repository import InMemoryAiPlatformRepository
class UnavailableObjectStorage:
def ensure_ready(self) -> None:
raise RuntimeError("object storage unavailable")
monkeypatch.setattr(main, "build_object_storage", lambda **kwargs: UnavailableObjectStorage())
with pytest.raises(RuntimeError, match="object storage unavailable"):
main.create_app(
AiPlatformSettings(_env_file=None, environment="test", repository_backend="memory"),
repository=InMemoryAiPlatformRepository(),
)