From 08704f91a40eed0d778f0574ca81b1d2c327e65e Mon Sep 17 00:00:00 2001 From: Sam Welborn Date: Mon, 29 Jun 2026 13:10:18 -0400 Subject: [PATCH 1/4] process tools to get local ports/hostname --- src/fastcache_api/process.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/src/fastcache_api/process.py b/src/fastcache_api/process.py index 63778ba..a50b1bc 100644 --- a/src/fastcache_api/process.py +++ b/src/fastcache_api/process.py @@ -2,6 +2,7 @@ import logging import socket import subprocess +from collections.abc import Iterable from functools import lru_cache from uuid import UUID @@ -20,6 +21,28 @@ def local_hostnames() -> set[str]: return {name.lower() for name in names if name} +def canonical_hostname() -> str: + """This host's preferred public name for cache ZMQ URIs.""" + return (socket.getfqdn() or socket.gethostname()).lower() + + +def ports_in_use(configs: Iterable[CacheConfig]) -> set[int]: + used: set[int] = set() + for config in configs: + for uri in (config.pull_uri, config.push_uri): + if uri.port is not None: + used.add(uri.port) + return used + + +def allocate_port_pair(in_use: set[int], start: int, end: int) -> tuple[int, int]: + for pull in range(start, end + 1, 2): + push = pull + 1 + if push <= end and pull not in in_use and push not in in_use: + return pull, push + raise RuntimeError(f"no free cache port pair in range [{start}, {end}]") + + def start_cache(cache_id: UUID, config: CacheConfig) -> CacheProcess: run_dir = settings.RUN_DIR / str(cache_id) run_dir.mkdir(parents=True, exist_ok=True) From f869cfc70e68c7c434461e2123f3247e33b82d20 Mon Sep 17 00:00:00 2001 From: Sam Welborn Date: Mon, 29 Jun 2026 13:12:01 -0400 Subject: [PATCH 2/4] add cache port pool to config --- src/fastcache_api/config.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/fastcache_api/config.py b/src/fastcache_api/config.py index 174adc8..4523e42 100644 --- a/src/fastcache_api/config.py +++ b/src/fastcache_api/config.py @@ -45,6 +45,9 @@ class Settings(BaseSettings): OIDC_JWKS_URI: str | None = None OIDC_AUDIENCES: Annotated[list[str], BeforeValidator(parse_comma_list)] = [] + CACHE_PORT_START: int = 30000 + CACHE_PORT_END: int = 30100 + # Cache process management # Path to the fastcache executable (overridable; defaults to PATH lookup). FASTCACHE_BINARY: Path = Path("lclstream-fastcache") From 130a8d1a236d3c7b888d74badcc57dc567d8c102 Mon Sep 17 00:00:00 2001 From: Sam Welborn Date: Mon, 29 Jun 2026 13:29:37 -0400 Subject: [PATCH 3/4] cacherequest loses pull/push ports, adds requested_by . --- src/fastcache_api/models.py | 7 +++---- src/fastcache_api/routes/cache.py | 8 ++++---- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/src/fastcache_api/models.py b/src/fastcache_api/models.py index 5206533..ee77180 100644 --- a/src/fastcache_api/models.py +++ b/src/fastcache_api/models.py @@ -62,10 +62,9 @@ def from_cache_config(cls, config: CacheConfig) -> FastcacheConfig: class CacheRequest(BaseModel): transfer_id: str - hostname: str - # later relax to none - pull_port: int - push_port: int + # Human who initiated the transfer upstream (bearer token is a shared + # service identity, so attribution must travel in the request body). + requested_by: str class CachePublic(BaseModel): diff --git a/src/fastcache_api/routes/cache.py b/src/fastcache_api/routes/cache.py index 3663463..2065702 100644 --- a/src/fastcache_api/routes/cache.py +++ b/src/fastcache_api/routes/cache.py @@ -64,9 +64,9 @@ async def create_cache( ) config = CacheConfig( - hostname=req.hostname, - pull_uri=f"tcp://{req.hostname}:{req.pull_port}", - push_uri=f"tcp://{req.hostname}:{req.push_port}", + hostname=hostname, + pull_uri=f"tcp://{hostname}:{pull_port}", + push_uri=f"tcp://{hostname}:{push_port}", ) cache_id = uuid4() @@ -83,7 +83,7 @@ async def create_cache( transfer_id=req.transfer_id, pid=proc.pid, create_time=proc.create_time, - user=user.email, + user=req.requested_by, status=CacheStatus.active, log_path=str(proc.log_path), config=config.model_dump(mode="json"), From 4c89a5c79fddc57323ba471ee21d8dd5a0710066 Mon Sep 17 00:00:00 2001 From: Sam Welborn Date: Mon, 29 Jun 2026 13:31:42 -0400 Subject: [PATCH 4/4] search for used ports in local db first --- src/fastcache_api/process.py | 8 ------- src/fastcache_api/routes/cache.py | 36 ++++++++++++++++++++++++------- 2 files changed, 28 insertions(+), 16 deletions(-) diff --git a/src/fastcache_api/process.py b/src/fastcache_api/process.py index a50b1bc..82c1624 100644 --- a/src/fastcache_api/process.py +++ b/src/fastcache_api/process.py @@ -3,7 +3,6 @@ import socket import subprocess from collections.abc import Iterable -from functools import lru_cache from uuid import UUID import psutil @@ -14,13 +13,6 @@ logger = logging.getLogger(__name__) -@lru_cache(maxsize=1) -def local_hostnames() -> set[str]: - names = {socket.gethostname(), socket.getfqdn()} - names |= {name.split(".", 1)[0] for name in names} - return {name.lower() for name in names if name} - - def canonical_hostname() -> str: """This host's preferred public name for cache ZMQ URIs.""" return (socket.getfqdn() or socket.gethostname()).lower() diff --git a/src/fastcache_api/routes/cache.py b/src/fastcache_api/routes/cache.py index 2065702..f9f1843 100644 --- a/src/fastcache_api/routes/cache.py +++ b/src/fastcache_api/routes/cache.py @@ -16,7 +16,13 @@ CachesPublic, CacheStatus, ) -from ..process import local_hostnames, start_cache, stop_cache +from ..process import ( + allocate_port_pair, + canonical_hostname, + ports_in_use, + start_cache, + stop_cache, +) from ..tables import Cache router = APIRouter( @@ -54,14 +60,28 @@ async def create_cache( user: Annotated[TokenPayload, Depends(require_user)], session: Annotated[AsyncSession, Depends(get_session)], ): - if req.hostname.lower() not in local_hostnames(): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=( - f"Cache must run on this api server's host; request specified " - f"'{req.hostname}'" - ), + # The cache always runs as a local subprocess of this api server, so the + # ZMQ URIs are published under this host's canonical (FQDN) name. + hostname = canonical_hostname() + + # Derive the ports already bound by live caches and allocate the next free pair. + active = await session.execute( + select(Cache).where( + Cache.state.in_([s.value for s in CacheState if not s.is_final()]) ) + ) + in_use = ports_in_use( + CacheConfig.model_validate(cache.config) for cache in active.scalars().all() + ) + try: + pull_port, push_port = allocate_port_pair( + in_use, settings.CACHE_PORT_START, settings.CACHE_PORT_END + ) + except RuntimeError as exc: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=str(exc), + ) from exc config = CacheConfig( hostname=hostname,