diff --git a/client/src/lclstream_api_client/_generated/api/caches_api.py b/client/src/lclstream_api_client/_generated/api/caches_api.py index 9fd2bca..4e25777 100644 --- a/client/src/lclstream_api_client/_generated/api/caches_api.py +++ b/client/src/lclstream_api_client/_generated/api/caches_api.py @@ -15,8 +15,7 @@ from typing import Any, Dict, List, Optional, Tuple, Union from typing_extensions import Annotated -from pydantic import StrictBool, StrictStr -from typing import Optional +from pydantic import StrictStr from uuid import UUID from lclstream_api_client._generated.models.caches_public import CachesPublic from lclstream_api_client._generated.models.message import Message @@ -409,7 +408,6 @@ def _get_caches_caches_get_serialize( async def shutdown_cache_caches_cache_id_delete( self, cache_id: UUID, - force: Optional[StrictBool] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], @@ -428,8 +426,6 @@ async def shutdown_cache_caches_cache_id_delete( :param cache_id: (required) :type cache_id: UUID - :param force: - :type force: bool :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of @@ -454,7 +450,6 @@ async def shutdown_cache_caches_cache_id_delete( _param = self._shutdown_cache_caches_cache_id_delete_serialize( cache_id=cache_id, - force=force, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, @@ -481,7 +476,6 @@ async def shutdown_cache_caches_cache_id_delete( async def shutdown_cache_caches_cache_id_delete_with_http_info( self, cache_id: UUID, - force: Optional[StrictBool] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], @@ -500,8 +494,6 @@ async def shutdown_cache_caches_cache_id_delete_with_http_info( :param cache_id: (required) :type cache_id: UUID - :param force: - :type force: bool :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of @@ -526,7 +518,6 @@ async def shutdown_cache_caches_cache_id_delete_with_http_info( _param = self._shutdown_cache_caches_cache_id_delete_serialize( cache_id=cache_id, - force=force, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, @@ -553,7 +544,6 @@ async def shutdown_cache_caches_cache_id_delete_with_http_info( async def shutdown_cache_caches_cache_id_delete_without_preload_content( self, cache_id: UUID, - force: Optional[StrictBool] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], @@ -572,8 +562,6 @@ async def shutdown_cache_caches_cache_id_delete_without_preload_content( :param cache_id: (required) :type cache_id: UUID - :param force: - :type force: bool :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of @@ -598,7 +586,6 @@ async def shutdown_cache_caches_cache_id_delete_without_preload_content( _param = self._shutdown_cache_caches_cache_id_delete_serialize( cache_id=cache_id, - force=force, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, @@ -621,7 +608,6 @@ async def shutdown_cache_caches_cache_id_delete_without_preload_content( def shutdown_cache_caches_cache_id_delete_sync( self, cache_id: UUID, - force: Optional[StrictBool] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], @@ -643,7 +629,6 @@ def shutdown_cache_caches_cache_id_delete_sync( return run_sync( self.shutdown_cache_caches_cache_id_delete( cache_id=cache_id, - force=force, _request_timeout=_request_timeout, _request_auth=_request_auth, _content_type=_content_type, @@ -657,7 +642,6 @@ def shutdown_cache_caches_cache_id_delete_sync( def shutdown_cache_caches_cache_id_delete_sync_with_http_info( self, cache_id: UUID, - force: Optional[StrictBool] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], @@ -679,7 +663,6 @@ def shutdown_cache_caches_cache_id_delete_sync_with_http_info( return run_sync( self.shutdown_cache_caches_cache_id_delete_with_http_info( cache_id=cache_id, - force=force, _request_timeout=_request_timeout, _request_auth=_request_auth, _content_type=_content_type, @@ -693,7 +676,6 @@ def shutdown_cache_caches_cache_id_delete_sync_with_http_info( def shutdown_cache_caches_cache_id_delete_sync_without_preload_content( self, cache_id: UUID, - force: Optional[StrictBool] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], @@ -715,7 +697,6 @@ def shutdown_cache_caches_cache_id_delete_sync_without_preload_content( return run_sync( self.shutdown_cache_caches_cache_id_delete_without_preload_content( cache_id=cache_id, - force=force, _request_timeout=_request_timeout, _request_auth=_request_auth, _content_type=_content_type, @@ -728,7 +709,6 @@ def shutdown_cache_caches_cache_id_delete_sync_without_preload_content( def _shutdown_cache_caches_cache_id_delete_serialize( self, cache_id, - force, _request_auth, _content_type, _headers, @@ -753,10 +733,6 @@ def _shutdown_cache_caches_cache_id_delete_serialize( if cache_id is not None: _path_params['cache_id'] = cache_id # process the query parameters - if force is not None: - - _query_params.append(('force', force)) - # process the header parameters # process the form parameters # process the body parameter diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 96c0472..1b8f6ba 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -1171,12 +1171,7 @@ export type ShutdownCacheCachesCacheIdDeleteData = { */ cache_id: string; }; - query?: { - /** - * Force - */ - force?: boolean; - }; + query?: never; url: '/caches/{cache_id}'; }; diff --git a/frontend/src/client/zod.gen.ts b/frontend/src/client/zod.gen.ts index 1179d31..fe6baa1 100644 --- a/frontend/src/client/zod.gen.ts +++ b/frontend/src/client/zod.gen.ts @@ -731,9 +731,7 @@ export const zShutdownCacheCachesCacheIdDeleteData = z.object({ path: z.object({ cache_id: z.uuid() }), - query: z.object({ - force: z.boolean().optional().default(false) - }).optional() + query: z.never().optional() }); /** diff --git a/frontend/src/components/Transfers/SharedCacheBadge.tsx b/frontend/src/components/Transfers/SharedCacheBadge.tsx index 09e4047..8eb4b90 100644 --- a/frontend/src/components/Transfers/SharedCacheBadge.tsx +++ b/frontend/src/components/Transfers/SharedCacheBadge.tsx @@ -36,12 +36,11 @@ export function SharedCacheBadge({ experiment }: { experiment: string }) { const cache = data?.data[0] if (!cache) return null - async function handleShutdown(force = false) { + async function handleShutdown() { if (!cache) return try { await shutdown.mutateAsync({ path: { cache_id: cache.id }, - query: force ? { force: true } : {}, }) toast.success("Cache shutdown requested") queryClient.invalidateQueries({ queryKey: query.queryKey }) @@ -54,7 +53,7 @@ export function SharedCacheBadge({ experiment }: { experiment: string }) { `${count} other active transfer(s) still use this cache. Shut it down anyway?`, ) ) { - await handleShutdown(true) + await handleShutdown() } return } diff --git a/openapi.json b/openapi.json index ff6a6d9..d393b1b 100644 --- a/openapi.json +++ b/openapi.json @@ -1724,16 +1724,6 @@ "title": "Cache Id", "type": "string" } - }, - { - "in": "query", - "name": "force", - "required": false, - "schema": { - "default": false, - "title": "Force", - "type": "boolean" - } } ], "responses": { diff --git a/src/lclstream_api/v2/exceptions.py b/src/lclstream_api/v2/exceptions.py index e36335e..5519e23 100644 --- a/src/lclstream_api/v2/exceptions.py +++ b/src/lclstream_api/v2/exceptions.py @@ -2,8 +2,6 @@ from fastapi.responses import JSONResponse from pydantic import BaseModel -from .models import CacheShutdownConflict - class NotFound(Exception): pass @@ -27,6 +25,11 @@ def __init__(self, active_transfer_count: int) -> None: ) +class CacheShutdownConflict(BaseModel): + message: str + active_transfer_count: int + + class InsufficientTokenLifetime(Exception): """The caller's token cannot cover the requested producer lifecycle.""" diff --git a/src/lclstream_api/v2/models.py b/src/lclstream_api/v2/models.py index 27164dd..b1a423b 100644 --- a/src/lclstream_api/v2/models.py +++ b/src/lclstream_api/v2/models.py @@ -126,7 +126,3 @@ class CacheStatusPublic(BaseModel): class CachesPublic(BaseModel): data: list[CacheStatusPublic] - -class CacheShutdownConflict(BaseModel): - message: str - active_transfer_count: int diff --git a/src/lclstream_api/v2/routers/v1/cache.py b/src/lclstream_api/v2/routers/v1/cache.py index 403e435..2b45805 100644 --- a/src/lclstream_api/v2/routers/v1/cache.py +++ b/src/lclstream_api/v2/routers/v1/cache.py @@ -7,7 +7,8 @@ from ... import service from ...auth import CurrentUser from ...db import get_session -from ...models import CacheShutdownConflict, CachesPublic, Message +from ...exceptions import CacheShutdownConflict +from ...models import CachesPublic, Message router = APIRouter(prefix="/caches", tags=["caches"]) @@ -29,7 +30,7 @@ async def get_caches( responses={409: {"model": CacheShutdownConflict}}, ) async def shutdown_cache( - cache_id: UUID, session: SessionDep, user: CurrentUser, force: bool = False + cache_id: UUID, session: SessionDep, user: CurrentUser ) -> Message: - await service.shutdown_cache(session, cache_id, force=force) + await service.shutdown_cache(session, cache_id) return Message(message="Cache shutdown requested") diff --git a/src/lclstream_api/v2/service.py b/src/lclstream_api/v2/service.py index 97d954e..c5badd6 100644 --- a/src/lclstream_api/v2/service.py +++ b/src/lclstream_api/v2/service.py @@ -271,8 +271,8 @@ async def list_transfer_logs( async def list_caches_for_experiment( session: AsyncSession, experiment: str ) -> CachesPublic: - """The experiment's active shared cache, if any.""" - cache_id = await repo.find_latest_shared_transfer_cache(session, experiment) + """The experiment's active cache, if any.""" + cache_id = await repo.find_active_cache(session, experiment) if cache_id is None: return CachesPublic(data=[]) cache = await fastcache.client().get_cache(cache_id) @@ -281,11 +281,17 @@ async def list_caches_for_experiment( return CachesPublic(data=[CacheStatusPublic(id=cache.id, state=cache.state)]) -async def shutdown_cache( - session: AsyncSession, cache_id: UUID, *, force: bool = False -) -> None: - if not force: +async def shutdown_cache(session: AsyncSession, cache_id: UUID) -> None: + async with session.begin(): + cache = await repo.lock_active_cache(session, cache_id) + if cache is None: + raise NotFound(f"cache {cache_id} not found") count = await repo.count_active_transfers_by_cache(session, cache_id) if count > 0: raise CacheShutdownBlocked(count) - await fastcache.client().delete_cache(cache_id) + + # Keep the registry row locked across deletion. Provisioning either + # attached before this count (and blocks shutdown), or waits and sees + # retired_at after commit, so it cannot attach to a deleted cache. + await fastcache.client().delete_cache(cache_id) + await repo.retire_cache(session, cache, retired_at=datetime.now(UTC))