diff --git a/src/fastcache_api/models.py b/src/fastcache_api/models.py index ee77180..b7aa1dd 100644 --- a/src/fastcache_api/models.py +++ b/src/fastcache_api/models.py @@ -1,11 +1,12 @@ +import json from enum import StrEnum from pathlib import Path from uuid import UUID -from pydantic import BaseModel, ConfigDict +from pydantic import AnyUrl, BaseModel, ConfigDict -class CacheStatus(StrEnum): +class CacheState(StrEnum): new = "new" queued = "queued" active = "active" @@ -21,42 +22,30 @@ class CacheConfig(BaseModel): """Configuration document persisted as JSON on a cache row. Mirrors the cacheserver config (see config/default.json) plus the - resolved ZMQ bind URIs. + resolved ZMQ bind URIs. ``to_fastcache_json`` renders the on-disk config + the fastcache binary consumes (see src/config.cpp): the ZMQ URIs are + renamed to ``inurl``/``outurl`` and ``hostname`` is dropped. """ hostname: str - pull_uri: str - push_uri: str + pull_uri: AnyUrl + push_uri: AnyUrl type: int = 4 helper_threads: int = 0 - num_workers: int = 1 io_threads: int = 16 hwm: int = 10 + # Default to 2 minutes for now. + timeout: int = 120_000 verbose: bool = False - -class FastcacheConfig(BaseModel): - """On-disk config consumed by the fastcache binary (see src/config.cpp). - - We should adjust these fields later to better match the request schema. - """ - - inurl: str - outurl: str - type: int = 4 - helper_threads: int = 0 - num_workers: int = 1 - io_threads: int = 16 - hwm: int = 10 - verbose: bool = False - - @classmethod - def from_cache_config(cls, config: CacheConfig) -> FastcacheConfig: - cfg_dict = config.model_dump(mode="json", exclude={"pull_uri", "push_uri"}) - return cls( - inurl=config.pull_uri, - outurl=config.push_uri, - **cfg_dict, + def to_fastcache_json(self, indent: int = 2) -> str: + """Render the config consumed by the fastcache binary.""" + data = self.model_dump( + mode="json", exclude={"hostname", "pull_uri", "push_uri"} + ) + return json.dumps( + {"inurl": str(self.pull_uri), "outurl": str(self.push_uri), **data}, + indent=indent, ) @@ -73,7 +62,7 @@ class CachePublic(BaseModel): id: UUID transfer_id: str user: str - status: CacheStatus + state: CacheState log_path: Path config: CacheConfig diff --git a/src/fastcache_api/process.py b/src/fastcache_api/process.py index 82c1624..efd5572 100644 --- a/src/fastcache_api/process.py +++ b/src/fastcache_api/process.py @@ -8,7 +8,7 @@ import psutil from .config import settings -from .models import CacheConfig, CacheProcess, FastcacheConfig +from .models import CacheConfig, CacheProcess logger = logging.getLogger(__name__) @@ -40,9 +40,7 @@ def start_cache(cache_id: UUID, config: CacheConfig) -> CacheProcess: run_dir.mkdir(parents=True, exist_ok=True) config_path = run_dir / "config.json" - config_path.write_text( - FastcacheConfig.from_cache_config(config).model_dump_json(indent=2) - ) + config_path.write_text(config.to_fastcache_json()) log_path = (run_dir / "cache.log").resolve() with log_path.open("ab") as log_file: diff --git a/src/fastcache_api/reconcile.py b/src/fastcache_api/reconcile.py index d08cb08..3ac6470 100644 --- a/src/fastcache_api/reconcile.py +++ b/src/fastcache_api/reconcile.py @@ -5,26 +5,24 @@ from .config import settings from .db import SessionLocal -from .models import CacheStatus +from .models import CacheState from .process import is_alive from .tables import Cache logger = logging.getLogger(__name__) -# Statuses whose process should still be running, so must be checked vs reality. -_NON_FINAL = [s.value for s in CacheStatus if not s.is_final()] +# States whose process should still be running, so must be checked vs reality. +_NON_FINAL = [s.value for s in CacheState if not s.is_final()] async def sweep_dead_caches() -> int: """Mark non-final caches whose process is gone as failed; return how many. - Caches outlive the api server, so a row's status is a claim we verify against + Caches outlive the api server, so a row's state is a claim we verify against the live process. """ async with SessionLocal() as session: - result = await session.execute( - select(Cache).where(Cache.status.in_(_NON_FINAL)) - ) + result = await session.execute(select(Cache).where(Cache.state.in_(_NON_FINAL))) caches = result.scalars().all() stale = 0 @@ -36,7 +34,7 @@ async def sweep_dead_caches() -> int: cache.id, cache.pid, ) - cache.status = CacheStatus.failed + cache.state = CacheState.failed stale += 1 if stale: diff --git a/src/fastcache_api/routes/cache.py b/src/fastcache_api/routes/cache.py index f9f1843..d475130 100644 --- a/src/fastcache_api/routes/cache.py +++ b/src/fastcache_api/routes/cache.py @@ -14,7 +14,7 @@ CachePublic, CacheRequest, CachesPublic, - CacheStatus, + CacheState, ) from ..process import ( allocate_port_pair, @@ -104,7 +104,7 @@ async def create_cache( pid=proc.pid, create_time=proc.create_time, user=req.requested_by, - status=CacheStatus.active, + state=CacheState.active, log_path=str(proc.log_path), config=config.model_dump(mode="json"), ) @@ -137,7 +137,7 @@ async def shutdown_cache( status_code=status.HTTP_404_NOT_FOUND, detail="Cache not found" ) - cache.status = CacheStatus.canceled + cache.state = CacheState.canceled await session.commit() await session.refresh(cache) diff --git a/src/fastcache_api/tables.py b/src/fastcache_api/tables.py index 38c0336..da8b021 100644 --- a/src/fastcache_api/tables.py +++ b/src/fastcache_api/tables.py @@ -9,7 +9,7 @@ ) from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column -from .models import CacheStatus +from .models import CacheState class Base(DeclarativeBase): @@ -46,9 +46,9 @@ class Cache(DTMixin, Base): doc=("psutil process create_time at launch. We can key by pid/create_time."), ) user: Mapped[str] = mapped_column(doc="User that created this cache") - status: Mapped[str] = mapped_column( - default=CacheStatus.new, - doc="Lifecycle status; stored as the CacheStatus string value", + state: Mapped[str] = mapped_column( + default=CacheState.new, + doc="Lifecycle state; stored as the CacheState string value", ) log_path: Mapped[str] = mapped_column( doc="Absolute path to the cache process log file",