create ports locally to fastcache server

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")

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):

src/fastcache_api/process.py

@@ -2,7 +2,7 @@
 import logging
 import socket
 import subprocess
-from functools import lru_cache
+from collections.abc import Iterable
 from uuid import UUID
 
 import psutil
@@ -13,11 +13,26 @@
 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()
+
+
+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:

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,19 +60,33 @@ 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=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 +103,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"),