diff --git a/deploy/helm/sie-cluster/README.md b/deploy/helm/sie-cluster/README.md index 364612ccf..97205464d 100644 --- a/deploy/helm/sie-cluster/README.md +++ b/deploy/helm/sie-cluster/README.md @@ -738,7 +738,10 @@ Use one of these patterns deliberately: through `/v1/pools` with `queue_pool` set to that physical queue. Use a logical pool name that does not collide with the protected static queue-pool name, or target the static pool directly. SDK calls use - `gpu="/"`. + `gpu="/"`. A logical pool created with + `ttl_seconds: "never"` does not expire either, like a static queue pool, but + stays API-owned: it can be changed and deleted through `/v1/pools` and is held + to the pool API bounds below. Missing named pools intentionally do not fail open. Falling back from `pool=default,machineProfile=l4` to `pool=l4,machineProfile=l4` would cross the @@ -1383,7 +1386,9 @@ and `POST /v1/configs/resolve`. They accept the token in the Secret named by `gateway.auth.adminTokenSecretName` (key `gateway.auth.adminTokenSecretKey`, default `SIE_ADMIN_TOKEN`), which the chart passes to the gateway as `SIE_ADMIN_TOKEN`. When it is empty, those requests get `403` (`Admin token not -configured`). The chart does not generate this token, and it must be separate +configured`). The gateway also accepts the admin token wherever it accepts a +client token, so an engineer who manages pools can use one key to create a +pool and send it traffic. The chart does not generate this token, and it must be separate from the sie-config tokens: the render fails when it names the Secret and key of the sie-config admin token, which would give every gateway pod catalog write access, or of the read token, which every worker sidecar holds. @@ -1394,8 +1399,8 @@ write access, or of the read token, which every worker sidecar holds. - **sie-config read token:** sie-config, every gateway pod, and every worker sidecar. It can read the whole catalog, including every model config, but cannot write. -- **Gateway admin token (when set):** every gateway pod and the tooling that - manages pools. +- **Gateway admin token (when set):** every gateway pod and the tooling and + engineers that manage pools. It also works as a client token. - **Gateway client token (when set or generated):** API clients, every gateway pod, and every worker sidecar, which presents it when it calls the gateway. See [Securing the cluster](#securing-the-cluster). @@ -1935,20 +1940,43 @@ an admin token is configured. With gateway token auth, pool create, renew, and delete also require the gateway admin token (see [Gateway admin token](#gateway-admin-token)). -Independently of auth, the gateway bounds API-created pools: + +The chart sets the envelope that pools work inside: the machine profiles and +lanes in `workers.pools`, and each lane's KEDA `maxReplicas`. Within it, API +callers create pools. Independently of auth, the gateway bounds API-created +pools: - the warm floor (`minimum_worker_count`, which applies to each of the pool's machine profiles) and the `gpus` requirement may each add up to at most 4 workers per pool by default, and a pool's active lease keeps at most its - requirement warm; -- the lease TTL is capped at 3600 s by default; -- live API-created pools are capped at 64 by default. Each gateway replica - checks the pools it knows, including ones replicated from other replicas, - so concurrent creates on different replicas can briefly exceed the cap; it - is not a strict cluster-wide limit. + requirement warm. This bounds only the always-on floor: KEDA still scales + the lane above it on queue depth and pending demand, up to the lane's + `maxReplicas`, and a pool without `gpu_caps` is assigned every worker on its + lanes; +- a pool's lease (`ttl_seconds`) is capped at 30 days by default with gateway + auth on, and at one hour with it off, so one unauthenticated request cannot + hold warm capacity for long. A pool created with `ttl_seconds: "never"` has + no lease and lives until it is deleted; the gateway accepts it only with + gateway auth on, where the request carries the admin token, and refuses it + with `403` otherwise; +- live API-created pools are capped at 64 by default, pools that never expire + included. Each gateway replica checks the pools it knows, including ones + replicated from other replicas, so concurrent creates on different replicas + can briefly exceed the cap; it is not a strict cluster-wide limit. Tune them with `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT`, -`SIE_GATEWAY_POOL_MAX_TTL_S`, and `SIE_GATEWAY_MAX_POOLS` in `gateway.extraEnv`. +`SIE_GATEWAY_POOL_MAX_TTL_S`, and `SIE_GATEWAY_MAX_POOLS` in `gateway.extraEnv`; +an explicit `SIE_GATEWAY_POOL_MAX_TTL_S` overrides either TTL default. + +Give services that only send traffic a client token rather than the admin +token: the admin token can also change and delete every pool. + +> **Upgrade note:** finish rolling out the gateway before you create a pool +> with `ttl_seconds: "never"`, change an existing pool to it, or give a pool a +> lease longer than one hour. A gateway replica from an earlier release cannot +> read a pool that never expires, keeps its older copy of the pool, and caps +> leases at one hour, so it can delete such a pool when that copy's lease runs +> out, which removes it from every replica. > **Upgrade note (breaking):** `values-aws.yaml`, `values-gke.yaml`, and > `values-aks.yaml` used to enable a host-less, TLS-less Ingress in front of an diff --git a/deploy/helm/sie-cluster/values.yaml b/deploy/helm/sie-cluster/values.yaml index b72af4176..2fd513e74 100644 --- a/deploy/helm/sie-cluster/values.yaml +++ b/deploy/helm/sie-cluster/values.yaml @@ -358,9 +358,9 @@ gateway: # -- Existing Secret with the gateway admin token (key: # adminTokenSecretKey). With mode token or static, the gateway requires it # on POST, PUT, and DELETE under /v1/pools, /v1/admin, and /v1/configs; - # when empty, those requests get 403. It must not name the Secret and key - # of either sie-config token or the gateway client token (the render - # fails). + # when empty, those requests get 403. The gateway also accepts it wherever + # it accepts the client token. It must not name the Secret and key of + # either sie-config token or the gateway client token (the render fails). adminTokenSecretName: "" adminTokenSecretKey: "SIE_ADMIN_TOKEN" @@ -1823,7 +1823,9 @@ queueRouting: # lowercase before gateway storage/routing. Use this when # poolAdmission.enabled=true and a worker group sets a named non-default # workers.pools..queuePool but the pool is static/deploy-owned rather - # than created through the expiring /v1/pools API. + # than created through the /v1/pools API. They report ttl_seconds "never", + # like an API pool created to never expire, but stay operator-owned: the API + # cannot change or delete them and the pool API limits do not apply. # # Example: # staticQueuePools: diff --git a/packages/sie_gateway/README.md b/packages/sie_gateway/README.md index ecf28dd0b..6f374f918 100644 --- a/packages/sie_gateway/README.md +++ b/packages/sie_gateway/README.md @@ -12,7 +12,7 @@ See [`docs/architecture-guide.md`](docs/architecture-guide.md) for the authorita - **Worker discovery** — static URLs or Kubernetes service endpoints - **Health monitoring** — WebSocket streaming or NATS heartbeats - **Queue transport** — JetStream work publish, reply inbox collection, backpressure handling, and DLQ republish -- **Pool management** — named pools with TTLs, minimum worker counts, and Kubernetes-backed coordination +- **Pool management** — named pools with leases of up to 30 days or no expiry (with gateway auth on), warm floors that the lane autoscales above, and Kubernetes-backed coordination - **Model registry** — filesystem seed + background-retried snapshot from `sie-config` (`GET /v1/configs/export`) + live NATS deltas + periodic `GET /v1/configs/epoch` drift detection; the gateway is read-only - **Config write cutover** — `POST /v1/configs/models` is not registered on the gateway. Requests receive `405 Method Not Allowed` from axum's default router. Writes belong to the control plane at `SIE_CONFIG_SERVICE_URL` - **Worker-ack readiness** — `GET /v1/configs/models/{id}/status` reports per-replica `bundle_config_hash` acknowledgement plus the local `config_epoch` for admin tooling polling after a `sie-config` write @@ -95,7 +95,7 @@ Each `--flag` above has a matching `SIE_*` environment variable (see next sectio | `SIE_AUTH_MODE` | `none` | Auth mode for inbound requests: `none` disables, `token` (alias `static`) enforces. Values are case-sensitive. An unknown value, including `sie-server`'s `auto`, refuses every request with `500` except the probes and the API description (`/healthz`, `/readyz`, `/openapi.json`, `/docs`), and `main` logs a startup error naming the bad value. While it is unset on a non-loopback bind and the configuration is otherwise valid, `main` logs a warning: from 0.13.0 the gateway refuses to start that way | | `SIE_AUTH_TOKENS` | | CSV of valid tokens for inference and pool/config read endpoints, presented as `Authorization: Bearer ` or `x-api-key: ` (the first `Authorization` value decides when it carries a token, and `x-api-key` is then ignored; when that value is empty or blank, `x-api-key` decides; an unreadable first value is refused, and `Bearer` alone counts as the token `Bearer`). Later `Authorization` values are ignored, as before. A request that passes authentication with a repeated `Authorization` header whose values carry different tokens, include one the gateway cannot read, or carry a token only after the first value logs a warning once per process (a refused request never logs it); from 0.13.0 such a request gets `401`. Repeated `x-api-key` values that carry different tokens get `401`. If unset, the singular `SIE_AUTH_TOKEN` is used as a fallback. When auth is enabled and this list is empty, non-probe requests return `500` | | `SIE_AUTH_TOKEN` | | Singular alias for `SIE_AUTH_TOKENS` (fallback only; prefer the plural form) | -| `SIE_ADMIN_TOKEN` | | Admin bearer token the gateway requires inbound for admin-gated mutations: `POST/PUT/DELETE` on `/v1/configs/*`, `/v1/admin/*`, `/v1/pools/*`. If empty and an inbound request targets one of those paths, the middleware fails closed with `403`. Sent to `sie-config` only when `SIE_CONFIG_SERVICE_TOKEN` is unset (deprecated fallback) | +| `SIE_ADMIN_TOKEN` | | Admin bearer token the gateway requires inbound for admin-gated mutations: `POST/PUT/DELETE` on `/v1/configs/*`, `/v1/admin/*`, `/v1/pools/*`. If empty and an inbound request targets one of those paths, the middleware fails closed with `403`. The admin token is also accepted wherever an `SIE_AUTH_TOKENS` token is, so one key can create a pool and send it traffic; give services that only send traffic an `SIE_AUTH_TOKENS` token instead, since the admin token can also change and delete every pool. Sent to `sie-config` only when `SIE_CONFIG_SERVICE_TOKEN` is unset (deprecated fallback) | | `SIE_CONFIG_SERVICE_TOKEN` | | Bearer token the gateway presents **as a client** on its `sie-config` reads (`/v1/configs/bundles`, `/export`, `/epoch`): `sie-config`'s read-scoped `SIE_CONFIG_READ_TOKEN`. When set, it decides and a blank value sends no token; when unset, the gateway falls back to `SIE_ADMIN_TOKEN` and logs a deprecation warning | | `SIE_WORKER_AUTH_TOKEN` | | Token the gateway presents as `Authorization: Bearer` when it dials a worker's `/ws/status` (`ws` health mode). Set it to the workers' `SIE_AUTH_TOKEN` when workers require a token. Like that variable it may list comma-separated tokens; the gateway presents the first. The dial uses plain `ws://`, so the token crosses the network unencrypted: keep the gateway-to-worker network trusted. Startup fails when it is not a valid header value | | `SIE_AUTH_EXEMPT_OPERATIONAL` | `false` | When `true`, `/`, `/health`, and `/ws/*` are exempt from auth (they expose worker URLs, queue depth, GPU inventory). `/healthz` and `/readyz` are always exempt (K8s probes carry no creds). Default is fail-closed | @@ -109,8 +109,8 @@ Each `--flag` above has a matching `SIE_*` environment variable (see next sectio | `SIE_GATEWAY_LANE_BACKPRESSURE_ENFORCE` | `false` | Act on the per-lane decision. Off = shadow mode: the decision is computed and recorded, and admission is governed by `SIE_GATEWAY_MAX_STREAM_PENDING` alone. On = a saturated lane gets a targeted 503 while other lanes on the same pool keep admitting | | `SIE_GATEWAY_DEFAULT_MAX_TOKENS` | `1024` | Output-token cap applied to `/v1/chat/completions` requests that omit both `max_completion_tokens` and `max_tokens`. OpenAI treats the field as optional, so the gateway defaults rather than rejecting — generic clients (Open WebUI) rely on this | | `SIE_GATEWAY_ENABLE_POOLS` | `false` | Enable pool management | -| `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT` | `4` | Largest number of workers one API-created pool may keep warm. `POST /v1/pools` rejects with `400 INVALID_REQUEST` a `minimum_worker_count` whose total across the pool's machine profiles (the floor applies to each profile) exceeds it, and `gpus` requirements summing to more than it. A pool stored before this budget existed is allotted it in machine-profile name order: it becomes Active once the allotted workers are available, its active lease keeps only those warm, and its warm floor is spread over its lanes (at least one worker each on as many lanes as the budget allows). Static Helm queue pools are exempt | -| `SIE_GATEWAY_POOL_MAX_TTL_S` | `3600` | Largest `ttl_seconds` `POST /v1/pools` accepts; larger values get `400 INVALID_REQUEST`. The lease of any stored pool, including one restored from Kubernetes, expires no later than this many seconds after its last renewal. `0` is ignored with a warning | +| `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT` | `4` | Largest number of workers one API-created pool may keep warm. It bounds only the always-on floor: KEDA still scales each lane above it on demand, up to the lane's Helm `maxReplicas`, and an uncapped pool is assigned every worker on its lanes. `POST /v1/pools` rejects with `400 INVALID_REQUEST` a `minimum_worker_count` whose total across the pool's machine profiles (the floor applies to each profile) exceeds it, and `gpus` requirements summing to more than it. A pool stored before this budget existed is allotted it in machine-profile name order: it becomes Active once the allotted workers are available, its active lease keeps only those warm, and its warm floor is spread over its lanes (at least one worker each on as many lanes as the budget allows). Static Helm queue pools are exempt | +| `SIE_GATEWAY_POOL_MAX_TTL_S` | `2592000` (30 days) with gateway auth on, `3600` with it off | Largest numeric `ttl_seconds` `POST /v1/pools` accepts; larger values and `0` get `400 INVALID_REQUEST`. The default depends on whether pool writes require the admin token, so that one unauthenticated request cannot hold warm capacity for long; setting the variable overrides either default. The lease of any stored pool with a lease, including one restored from Kubernetes, expires no later than this many seconds after its last renewal. `ttl_seconds: "never"` is outside this limit: it creates a pool without a lease that lives until it is deleted, and the gateway accepts it only with gateway auth on (`SIE_AUTH_MODE=token`, so the request carries the admin token); with auth off it is refused with `403 POOL_OPERATION_FORBIDDEN`. `0` is ignored with a warning | | `SIE_GATEWAY_MAX_POOLS` | `64` | Largest number of live API-created pools (the `default` pool and static Helm queue pools do not count). Creating one more gets `403 POOL_OPERATION_FORBIDDEN`; `0` disables API pool creation. Each replica checks the pools it knows, including ones replicated from other gateways, so concurrent creates on different replicas can briefly exceed it | | `SIE_GATEWAY_HOT_RELOAD` | `false` | Enable filesystem watcher for bundle/model directories | | `SIE_GATEWAY_WATCH_POLLING` | `false` | Use polling file-watcher instead of inotify/fsevents (alias: `SIE_GATEWAY_POLLING_WATCHER`). Useful on filesystems where native notifications are unreliable | @@ -182,7 +182,7 @@ The same gateway-owned OpenAPI contract is available at runtime via `GET /openap | GET | `/v1/pools` | List pools | | POST | `/v1/pools` | Create pool | | GET | `/v1/pools/{name}` | Get pool details | -| POST | `/v1/pools/{name}/renew` | Renew pool TTL | +| POST | `/v1/pools/{name}/renew` | Renew a pool's lease (a no-op for a pool that never expires) | | DELETE | `/v1/pools/{name}` | Delete pool (default pool protected) | ### Config API (read-only) diff --git a/packages/sie_gateway/docs/architecture-guide.md b/packages/sie_gateway/docs/architecture-guide.md index 556cd5568..6b87bc459 100644 --- a/packages/sie_gateway/docs/architecture-guide.md +++ b/packages/sie_gateway/docs/architecture-guide.md @@ -927,11 +927,11 @@ expiry; request cardinality therefore cannot create entries or Tokio tasks. Each physical queue pool gets its own JetStream stream (`WORK_POOL_{queue_pool}`) with subjects `sie.work.{queue_pool}.*.*.*`. Pool and queue-pool names are validated against `[A-Za-z0-9_-]` and canonicalized to lowercase at the API, gateway request, K8s restore/watch, and Helm render boundaries, so NATS subjects and KEDA labels do not split by case. Workers are deployed with `SIE_POOL={queue_pool}`, `SIE_MACHINE_PROFILE={profile}`, and `SIE_BUNDLE={bundle}`; they consume only from their concrete `sie.work.{queue_pool}.{profile}.{bundle}.*` lane. A logical API pool names the tenant/workload boundary and carries `PoolSpec.queue_pool` to select the physical queue lane; omitted `queue_pool` means `default`. Helm's baseline default puts enabled worker groups in `SIE_POOL=default`, so SDK calls can pass just the machine profile (`gpu="l4"`) for shared-cluster capacity, and API-created pools can borrow that same base capacity with `gpu="pool/l4"`. Dedicated queue deployments override Helm `queuePool` and must declare the same name under `queueRouting.staticQueuePools` before API pools can use it as a non-default `queue_pool`; callers can target the protected static pool directly or create a differently named logical pool backed by that queue, for example `{"name":"customer-acme-workload","queue_pool":"customer-acme"}` plus SDK `gpu="customer-acme-workload/l4"`. A cold `X-SIE-POOL` request without `X-SIE-MACHINE-PROFILE` records lane-specific `pending_demand` for every machine profile the pool can provision. Multi-profile pool demand therefore wakes all candidate lanes from zero instead of relying on an empty `machine_profile` label that no per-lane KEDA ScaledObject can observe. Registration does not clear demand: only a successful gateway work publish clears the exact target lane; other candidate lanes retain demand until work is published to them or the entry expires. This preserves scale-up pressure when registered workers are saturated. -A pool spec may also carry `minimum_worker_count` (a per-pool warm floor, default `0`). It keeps the first request from hitting a cold VM: every five seconds the gateway re-emits `sie.gateway.pool.warm_floor{pool,machine_profile,bundle}` for each backing queue lane the pool can name (clearing lanes that disappear), and a per-lane KEDA trigger reads the collector's Prometheus translation with `or vector(0)` so KEDA keeps machines warm. When multiple logical pools share one backing lane, the emitted value is the maximum requested floor for that physical lane, not a sum. Active lease capacity is labeled the same way. Default `0` emits nothing and leaves scale-from-zero unchanged. +A pool spec may also carry `minimum_worker_count` (a per-pool warm floor, default `0`). It keeps the first request from hitting a cold VM: every five seconds the gateway re-emits `sie.gateway.pool.warm_floor{pool,machine_profile,bundle}` for each backing queue lane the pool can name (clearing lanes that disappear), and a per-lane KEDA trigger reads the collector's Prometheus translation with `or vector(0)` so KEDA keeps machines warm. When multiple logical pools share one backing lane, the emitted value is the maximum requested floor for that physical lane, not a sum. Active lease capacity is labeled the same way. Default `0` emits nothing and leaves scale-from-zero unchanged. The floor is a minimum, not a size: KEDA's HPA takes the largest of the lane's triggers, so the queue-depth, pending-demand, and scale-worthy-rejection triggers scale the lane above the floor on demand, up to the lane's Helm `maxReplicas`, and back down to the floor when demand drops. A pool without `gpu_caps` is assigned every worker on its lanes as they arrive, so its capacity grows with the lane; `gpu_caps` is the per-pool assignment ceiling when one is wanted. A pool spec may also carry `pinned_models` (a per-pool set of always-loaded models, default empty), chosen from the models the gateway already tracks (`GET /v1/configs/models`); ids may be profile-qualified (`model-name:profile_name`, the registry's `{base}:{profile}` variant entries, with `model:default` folding to the base). The API validates each id against the registry and stores it canonicalized. Every 30 seconds the gateway emits the complete `sie.gateway.pool.pinned_model.loaded{pool,model}` snapshot through its canonical OTel facade: the value is `1` only when a healthy worker assigned to that logical pool reports the model loaded, and removed pools or pins are explicitly zeroed. Actually keeping the models loaded (worker preload) and lazy-load/LRU using spare capacity around the pinned set are worker-side and tracked separately. Default empty leaves lazy-loading unchanged. -Creating a usable custom pool has two supported shapes. For a dynamic logical pool over base capacity, call `POST /v1/pools` with the logical name and GPU requirements and omit `queue_pool`; it defaults to `default`, so no additional Helm worker group is required. Pool `name` is user/dynamic. Non-default `queue_pool` is infra/admin: it must name a Helm-rendered `queuePool`/`SIE_POOL` backing lane declared under `queueRouting.staticQueuePools`. For a dedicated physical queue, first add that worker group to Helm values and declare the same name under `queueRouting.staticQueuePools`, then create logical pools with `queue_pool` set to that queue. Lease-owned dynamic pools expire after their TTL unless renewed with `POST /v1/pools/{name}/renew`. The gateway bounds API-created pools independently of auth: a `minimum_worker_count` whose total across the pool's machine profiles, or `gpus` requirements whose sum, exceeds `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT` (default 4) and `ttl_seconds` above `SIE_GATEWAY_POOL_MAX_TTL_S` (default 3600) are rejected with `400 INVALID_REQUEST`, and a create beyond `SIE_GATEWAY_MAX_POOLS` live API-created pools (default 64) is rejected with `403 POOL_OPERATION_FORBIDDEN`. Pools restored from Kubernetes or applied from another replica are held to the same TTL and warm-floor caps when their lease expiry and KEDA warm floor are computed. An API-created pool's active lease keeps at most its `gpus` requirement per machine profile warm, and at most `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT` in total, however many workers it is assigned; a stored pool's per-lane warm floor is capped so its total fits the same budget. The live-pool cap is checked by the replica serving the create against every pool it knows, so concurrent creates on different replicas can briefly exceed it. `POST /v1/pools` refuses the name `default` with `403 POOL_OPERATION_FORBIDDEN`. The `default` pool and static Helm queue pools are operator-owned and exempt. Clients target either shape with an explicit `X-SIE-POOL` header, for example `X-SIE-POOL: customer-acme`, or SDK `gpu="customer-acme/l4"`. The `default` pool is protected and cannot be deleted. Missing named pools intentionally fail closed so capped/dynamic pool isolation is not weakened by a silent fallback. +Creating a usable custom pool has two supported shapes. For a dynamic logical pool over base capacity, call `POST /v1/pools` with the logical name and GPU requirements and omit `queue_pool`; it defaults to `default`, so no additional Helm worker group is required. Pool `name` is user/dynamic. Non-default `queue_pool` is infra/admin: it must name a Helm-rendered `queuePool`/`SIE_POOL` backing lane declared under `queueRouting.staticQueuePools`. For a dedicated physical queue, first add that worker group to Helm values and declare the same name under `queueRouting.staticQueuePools`, then create logical pools with `queue_pool` set to that queue. Every pool has one lifetime rule, set by `PoolSpec.ttl_seconds`: a number of seconds is a lease that expires that long after the pool's last renewal; omitted (`null`) is the gateway's default 1200 s lease; `"never"` is no lease at all, so the pool lives until it is deleted. The `default` pool and static Helm queue pools carry `"never"` in their spec, so expiry, renewal, and the Kubernetes Lease follow the spec alone, with no per-kind special case: a pool with a lease gets a `coordination.k8s.io` Lease for crash-safe expiry, and a pool without one gets none (a leftover Lease is dropped rather than allowed to collect it). `POST /v1/pools/{name}/renew` restarts a lease and is a harmless no-op on a pool that never expires; the SDKs renew leased pools every 60 s while the client runs and do not renew pools that never expire. Changing only a pool's `ttl_seconds` keeps its assigned workers, and an update that omits `ttl_seconds` keeps the pool's current lifetime, so a client that re-posts a pool at startup cannot shorten one an admin made permanent; send a number to give it a lease again. The gateway accepts `"never"` only with gateway auth on (`SIE_AUTH_MODE=token`), where every pool write carries the admin token; with auth off it refuses it with `403 POOL_OPERATION_FORBIDDEN`, so one unauthenticated request cannot hold warm capacity indefinitely. A stored pool that never expires is kept by every replica, including one restored from Kubernetes. The gateway bounds API-created pools independently of auth: a `minimum_worker_count` whose total across the pool's machine profiles, or `gpus` requirements whose sum, exceeds `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT` (default 4), a numeric `ttl_seconds` above `SIE_GATEWAY_POOL_MAX_TTL_S` (by default 2592000, 30 days, with gateway auth on and 3600 with it off), and a `ttl_seconds` of `0` are rejected with `400 INVALID_REQUEST`, and a create beyond `SIE_GATEWAY_MAX_POOLS` live API-created pools (default 64, pools that never expire included) is rejected with `403 POOL_OPERATION_FORBIDDEN`. Pools restored from Kubernetes or applied from another replica are held to the same TTL and warm-floor caps when their lease expiry and KEDA warm floor are computed. An API-created pool's active lease keeps at most its `gpus` requirement per machine profile warm, and at most `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT` in total, however many workers it is assigned; a stored pool's per-lane warm floor is capped so its total fits the same budget. The live-pool cap is checked by the replica serving the create against every pool it knows, so concurrent creates on different replicas can briefly exceed it. `POST /v1/pools` refuses the name `default` with `403 POOL_OPERATION_FORBIDDEN`. The `default` pool and static Helm queue pools are operator-owned and exempt. Clients target either shape with an explicit `X-SIE-POOL` header, for example `X-SIE-POOL: customer-acme`, or SDK `gpu="customer-acme/l4"`. The `default` pool is protected and cannot be deleted. Missing named pools intentionally fail closed so capped/dynamic pool isolation is not weakened by a silent fallback. Model configs registered via the control plane (`sie-config`) may carry a top-level `pool` field. That field is model routing/config assignment: omitted or empty means `default`, and a request that omits `X-SIE-POOL` defaults to the model's assigned pool. It does not create workers, KEDA ScaledObjects, or runtime pool leases. Worker counts, machine profiles, bundles, and autoscaling lanes are still Helm-driven physical capacity; `/v1/pools` and `queueRouting.staticQueuePools` only create or renew runtime admission/capacity state, optionally backed by existing `default` capacity through `PoolSpec.queue_pool`. In the Kubernetes/Helm composition, workers filter materialized model configs by their physical `SIE_POOL`, so a model config for a dedicated queue normally uses the physical queue-pool name; gateway requests may still target a logical pool whose `queue_pool` resolves to that physical name. A default-backed logical pool should not use a non-default model-config `pool` unless a matching dedicated worker queue exists. The managed Modal composition has one governed exception: a deployment-owned diagnostic admission queue may reuse the complete `default` catalog in a separate worker app. Dispatch stays on that physical queue, but bundle-hash and immutable-revision evidence are atomically scoped to the requested model's declared catalog pool. A known catalog model with a missing scoped hash fails closed rather than enabling the empty-hash wildcard; unknown sealed compatibility remains separately bounded. diff --git a/packages/sie_gateway/openapi.json b/packages/sie_gateway/openapi.json index 70418f705..7548db975 100644 --- a/packages/sie_gateway/openapi.json +++ b/packages/sie_gateway/openapi.json @@ -1134,12 +1134,14 @@ ] }, "ttl_seconds": { - "description": "Lease TTL in seconds; omit for the gateway default. Values above the\ngateway limit (`SIE_GATEWAY_POOL_MAX_TTL_S`, default 3600) are rejected.", - "format": "int64", - "minimum": 0, - "type": [ - "integer", - "null" + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "#/components/schemas/PoolTtl", + "description": "How long the pool lives after its last renewal. Omit for the gateway\ndefault lease (1200 s), which SDK clients renew while they run. A\nnumber of seconds sets the lease; values above the gateway limit\n(`SIE_GATEWAY_POOL_MAX_TTL_S`; by default 2592000, 30 days, with\ngateway auth on and 3600 without) and `0` are rejected with `400`. `\"never\"` creates a pool that lives until it is\ndeleted and needs no renewal; it requires gateway auth\n(`SIE_AUTH_MODE=token`, so the request carries the admin token) and is\nrefused with `403` otherwise." + } ] } }, @@ -3733,11 +3735,14 @@ "type": "string" }, "ttl_seconds": { - "format": "int64", - "minimum": 0, - "type": [ - "integer", - "null" + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "#/components/schemas/PoolTtl", + "description": "How long the pool lives after its last renewal: whole seconds, or\n`\"never\"` for a pool that lives until it is deleted. `null` on an\nAPI-created pool means the gateway's default lease. The `default` pool\nand static Helm queue pools report `\"never\"`." + } ] } }, @@ -3779,6 +3784,24 @@ ], "type": "object" }, + "PoolTtl": { + "description": "How long a pool lives after its last renewal: a whole number of seconds, or \"never\".", + "oneOf": [ + { + "description": "The pool expires this many seconds after its last renewal.", + "format": "int64", + "minimum": 1, + "type": "integer" + }, + { + "description": "The pool never expires; it lives until it is deleted.", + "enum": [ + "never" + ], + "type": "string" + } + ] + }, "ProfileInfoWire": { "properties": { "is_default": { @@ -8217,7 +8240,7 @@ } } }, - "description": "Invalid pool request, including a TTL or warm floor above the gateway limit" + "description": "Invalid pool request, including a zero TTL or a TTL or warm floor above the gateway limit" }, "401": { "content": { @@ -8237,7 +8260,7 @@ } } }, - "description": "Admin token required for this mutation (or admin token not configured), the pool is named `default`, or the gateway's limit on API-created pools is reached" + "description": "Admin token required for this mutation (or admin token not configured), the pool is named `default`, the gateway's limit on API-created pools is reached, or `ttl_seconds: \"never\"` was sent to a gateway without auth" }, "500": { "content": { @@ -8387,6 +8410,7 @@ }, "/v1/pools/{name}/renew": { "post": { + "description": "Restart the pool's lease. A pool created with `ttl_seconds: \"never\"` has no lease and needs no renewal; renewing it succeeds and changes nothing.", "operationId": "renew_pool", "parameters": [ { diff --git a/packages/sie_gateway/src/config.rs b/packages/sie_gateway/src/config.rs index ab3a453f6..7589b5aef 100644 --- a/packages/sie_gateway/src/config.rs +++ b/packages/sie_gateway/src/config.rs @@ -549,9 +549,13 @@ fn item_text_byte_limit_from_env() -> usize { /// Read the pool API bounds from `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT`, /// `SIE_GATEWAY_POOL_MAX_TTL_S`, and `SIE_GATEWAY_MAX_POOLS`. An unset value /// keeps that bound's default; an unparsable one keeps it with a warning. -pub fn pool_limits_from_env() -> PoolLimits { - let defaults = PoolLimits::default(); +/// `admin_gated` picks the defaults (see [`PoolLimits::for_pool_writes`]): it +/// must be true only when every `POST /v1/pools` is checked against the admin +/// token, which allows pools that never expire and a 30-day TTL limit. +pub fn pool_limits_from_env(admin_gated: bool) -> PoolLimits { + let defaults = PoolLimits::for_pool_writes(admin_gated); PoolLimits { + allow_no_expiry: defaults.allow_no_expiry, max_minimum_worker_count: env_pool_limit( "SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT", defaults.max_minimum_worker_count, @@ -1285,7 +1289,11 @@ mod tests { "SIE_GATEWAY_MAX_POOLS", ]; without_env(&keys, || { - assert_eq!(pool_limits_from_env(), PoolLimits::default()); + assert_eq!(pool_limits_from_env(false), PoolLimits::default()); + assert_eq!( + pool_limits_from_env(true), + PoolLimits::for_pool_writes(true) + ); }); with_env( &[ @@ -1295,11 +1303,12 @@ mod tests { ], || { assert_eq!( - pool_limits_from_env(), + pool_limits_from_env(false), PoolLimits { max_minimum_worker_count: 0, max_ttl_seconds: 86_400, max_pools: PoolLimits::default().max_pools, + allow_no_expiry: false, } ); }, @@ -1314,11 +1323,15 @@ mod tests { ("SIE_GATEWAY_MAX_POOLS", "0"), ], || { - let limits = pool_limits_from_env(); + let limits = pool_limits_from_env(false); assert_eq!( limits.max_ttl_seconds, PoolLimits::default().max_ttl_seconds ); + assert_eq!( + pool_limits_from_env(true).max_ttl_seconds, + PoolLimits::for_pool_writes(true).max_ttl_seconds + ); assert_eq!(limits.max_pools, 0); }, ); diff --git a/packages/sie_gateway/src/handlers/pools.rs b/packages/sie_gateway/src/handlers/pools.rs index 52a46ec24..310465c42 100644 --- a/packages/sie_gateway/src/handlers/pools.rs +++ b/packages/sie_gateway/src/handlers/pools.rs @@ -13,6 +13,7 @@ use crate::http_error::{code as err_code, json_detail}; use crate::server::AppState; use crate::state::model_registry::ModelRegistry; use crate::state::pool_manager::{PoolLimitError, DEFAULT_POOL_NAME}; +use crate::types::pool::PoolTtl; #[derive(Debug, Deserialize, ToSchema)] pub struct CreatePoolRequest { @@ -32,10 +33,16 @@ pub struct CreatePoolRequest { pub gpu_caps: HashMap, #[serde(default)] pub bundle: Option, - /// Lease TTL in seconds; omit for the gateway default. Values above the - /// gateway limit (`SIE_GATEWAY_POOL_MAX_TTL_S`, default 3600) are rejected. + /// How long the pool lives after its last renewal. Omit for the gateway + /// default lease (1200 s), which SDK clients renew while they run. A + /// number of seconds sets the lease; values above the gateway limit + /// (`SIE_GATEWAY_POOL_MAX_TTL_S`; by default 2592000, 30 days, with + /// gateway auth on and 3600 without) and `0` are rejected with `400`. `"never"` creates a pool that lives until it is + /// deleted and needs no renewal; it requires gateway auth + /// (`SIE_AUTH_MODE=token`, so the request carries the admin token) and is + /// refused with `403` otherwise. #[serde(default)] - pub ttl_seconds: Option, + pub ttl_seconds: Option, /// Per-pool warm floor (minimum machines kept warm via KEDA). Default 0 /// keeps scale-from-zero. It applies to each of the pool's machine /// profiles; a floor whose total across profiles exceeds the gateway limit @@ -104,8 +111,8 @@ fn canonicalize_pinned_models( request_body = CreatePoolRequest, responses( (status = 201, description = "Pool created, renewed, or updated", body = crate::types::pool::Pool), - (status = 400, description = "Invalid pool request, including a TTL or warm floor above the gateway limit", body = crate::openapi::StandardApiError), - (status = 403, description = "Admin token required for this mutation (or admin token not configured), the pool is named `default`, or the gateway's limit on API-created pools is reached", body = crate::openapi::StandardApiError) + (status = 400, description = "Invalid pool request, including a zero TTL or a TTL or warm floor above the gateway limit", body = crate::openapi::StandardApiError), + (status = 403, description = "Admin token required for this mutation (or admin token not configured), the pool is named `default`, the gateway's limit on API-created pools is reached, or `ttl_seconds: \"never\"` was sent to a gateway without auth", body = crate::openapi::StandardApiError) ) )] pub async fn create_pool( @@ -186,7 +193,7 @@ fn create_pool_error_response( error: &(dyn std::error::Error + Send + Sync + 'static), ) -> axum::response::Response { let (status, code) = match error.downcast_ref::() { - Some(PoolLimitError::TooManyPools { .. }) => { + Some(PoolLimitError::TooManyPools { .. } | PoolLimitError::NoExpiryRequiresAuth { .. }) => { (StatusCode::FORBIDDEN, err_code::POOL_OPERATION_FORBIDDEN) } _ => (StatusCode::BAD_REQUEST, err_code::INVALID_REQUEST), @@ -287,6 +294,7 @@ pub async fn delete_pool( path = "/v1/pools/{name}/renew", tag = "pools", params(("name" = String, Path, description = "Pool name")), + description = "Restart the pool's lease. A pool created with `ttl_seconds: \"never\"` has no lease and needs no renewal; renewing it succeeds and changes nothing.", responses( (status = 200, description = "Pool renewed", body = crate::openapi::MessageResponse), (status = 404, description = "Pool not found", body = crate::openapi::StandardApiError) diff --git a/packages/sie_gateway/src/main.rs b/packages/sie_gateway/src/main.rs index d3705413d..1641757ac 100644 --- a/packages/sie_gateway/src/main.rs +++ b/packages/sie_gateway/src/main.rs @@ -214,8 +214,9 @@ async fn run_server(cfg: Config) -> Result<(), Box> { let pools_enabled = config.enable_pools || config.use_kubernetes; // Set up pool manager (created early so on_worker_healthy callback can capture it) - let mut pm = PoolManager::new(config.configured_gpus.clone()) - .with_limits(config::pool_limits_from_env()); + let mut pm = PoolManager::new(config.configured_gpus.clone()).with_limits( + config::pool_limits_from_env(middleware::auth::is_auth_enabled(&config.auth_mode)), + ); let mut k8s_pool_backend: Option> = None; if config.use_kubernetes && pools_enabled { match state::k8s_pool_backend::K8sPoolBackend::new(&config.k8s_namespace, &router_id).await diff --git a/packages/sie_gateway/src/middleware/auth.rs b/packages/sie_gateway/src/middleware/auth.rs index 850d1465f..c62c64ae9 100644 --- a/packages/sie_gateway/src/middleware/auth.rs +++ b/packages/sie_gateway/src/middleware/auth.rs @@ -201,10 +201,14 @@ where return inner.call(req).await; } + // The admin token grants everything a client token does, so one + // credential can both manage pools and use them. let valid = config .auth_tokens .iter() - .any(|valid_token| constant_time_eq_str(&token, valid_token)); + .any(|valid_token| constant_time_eq_str(&token, valid_token)) + || (!config.admin_token.is_empty() + && constant_time_eq_str(&token, &config.admin_token)); if !valid { record_admission_rejection(&req, AdmissionOutcome::Unauthenticated); @@ -248,7 +252,8 @@ pub fn is_auth_enabled(mode: &str) -> bool { } /// Returns true for mutations that require the admin token. Reads on -/// these paths (including `GET`) are checked with the regular token. +/// these paths (including `GET`) are checked with the regular token, which +/// the admin token also satisfies. /// /// Admin-gated prefixes: /// - `/v1/config*` — reserved; config writes are currently 405. @@ -811,6 +816,43 @@ mod tests { ); } + #[tokio::test] + async fn middleware_admin_token_also_serves_client_routes() { + // One credential can create a pool and then send it traffic. + let cfg = cfg_for_middleware("token", vec!["user"], "admin", false); + for (method, path) in [ + (Method::POST, "/v1/encode/any"), + (Method::GET, "/v1/pools"), + (Method::GET, "/v1/pools/gpu-l4"), + ] { + let r = test_router(Arc::clone(&cfg)); + assert_eq!( + send(r, method.clone(), path, Some("admin")).await, + StatusCode::OK, + "{method} {path}" + ); + let r = test_router(Arc::clone(&cfg)); + assert_eq!( + send_headers(r, method.clone(), path, &[("x-api-key", "admin")]).await, + StatusCode::OK, + "{method} {path} via x-api-key" + ); + } + } + + #[tokio::test] + async fn middleware_unset_admin_token_adds_no_client_credential() { + let cfg = cfg_for_middleware("token", vec!["user"], "", false); + for bearer in [None, Some("admin"), Some("bogus")] { + let r = test_router(Arc::clone(&cfg)); + assert_eq!( + send(r, Method::POST, "/v1/encode/any", bearer).await, + StatusCode::UNAUTHORIZED, + "{bearer:?}" + ); + } + } + #[tokio::test] async fn middleware_admin_mutation_rejects_user_token() { let cfg = cfg_for_middleware("token", vec!["user"], "admin", false); diff --git a/packages/sie_gateway/src/server.rs b/packages/sie_gateway/src/server.rs index 9a1eaf556..5bd6da5a4 100644 --- a/packages/sie_gateway/src/server.rs +++ b/packages/sie_gateway/src/server.rs @@ -1243,7 +1243,7 @@ mod flat_404_tests { use crate::server::{create_router, AppState}; use crate::state::demand_tracker::DemandTracker; use crate::state::model_registry::ModelRegistry; - use crate::state::pool_manager::PoolManager; + use crate::state::pool_manager::{PoolLimits, PoolManager, DEFAULT_POOL_MAX_TTL_SECONDS}; use crate::state::worker_registry::WorkerRegistry; fn test_config(bundles_dir: &str, models_dir: &str) -> Config { @@ -1300,6 +1300,12 @@ mod flat_404_tests { // Returned tempdirs must outlive the router so they drop after the test. fn build_router() -> (Router, tempfile::TempDir, tempfile::TempDir) { + build_router_with_pool_limits(PoolLimits::default()) + } + + fn build_router_with_pool_limits( + limits: PoolLimits, + ) -> (Router, tempfile::TempDir, tempfile::TempDir) { let bundles_dir = tempfile::TempDir::new().unwrap(); let models_dir = tempfile::TempDir::new().unwrap(); let config = Arc::new(test_config( @@ -1314,7 +1320,7 @@ mod flat_404_tests { models_dir.path(), false, )), - pool_manager: Arc::new(PoolManager::new(Vec::new())), + pool_manager: Arc::new(PoolManager::new(Vec::new()).with_limits(limits)), work_publisher: None, lane_backlog_source: None, demand_tracker: Arc::new(DemandTracker::new(Default::default())), @@ -1397,11 +1403,12 @@ mod flat_404_tests { let (app, _bundles_dir, _models_dir) = build_router(); for body in [ serde_json::json!({"name": "bench", "gpus": {"l4": 1}, "minimum_worker_count": 99}), - serde_json::json!({"name": "bench", "gpus": {"l4": 1}, "ttl_seconds": 86_400}), + serde_json::json!({"name": "bench", "gpus": {"l4": 1}, "ttl_seconds": 3601}), + serde_json::json!({"name": "bench", "gpus": {"l4": 1}, "ttl_seconds": 0}), serde_json::json!({"name": "bench", "gpus": {"l4": 6}}), ] { - let (status, detail) = post_pool(&app, body).await; - assert_eq!(status, StatusCode::BAD_REQUEST); + let (status, detail) = post_pool(&app, body.clone()).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); assert_eq!(detail["detail"]["code"], "INVALID_REQUEST"); } let (status, _) = post_pool( @@ -1412,6 +1419,67 @@ mod flat_404_tests { assert_eq!(status, StatusCode::CREATED); } + #[tokio::test] + async fn test_create_pool_accepts_ttls_up_to_thirty_days_with_gateway_auth() { + let (app, _bundles_dir, _models_dir) = + build_router_with_pool_limits(PoolLimits::for_pool_writes(true)); + for ttl in [7 * 24 * 3600, DEFAULT_POOL_MAX_TTL_SECONDS] { + let (status, pool) = post_pool( + &app, + serde_json::json!({"name": "eval", "gpus": {"l4": 1}, "ttl_seconds": ttl}), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{ttl}"); + assert_eq!(pool["spec"]["ttl_seconds"], ttl); + } + } + + #[tokio::test] + async fn test_create_pool_that_never_expires_requires_gateway_auth() { + let never = serde_json::json!({"name": "rag", "gpus": {"l4": 1}, "ttl_seconds": "never"}); + + // `build_router` serves with auth off, where pool writes carry no + // admin token: a pool that never expires is refused. + let (app, _bundles_dir, _models_dir) = build_router(); + let (status, body) = post_pool(&app, never.clone()).await; + assert_eq!(status, StatusCode::FORBIDDEN); + assert_eq!(body["detail"]["code"], "POOL_OPERATION_FORBIDDEN"); + assert!(body["detail"]["message"] + .as_str() + .unwrap() + .contains("SIE_AUTH_MODE=token")); + + let (app, _bundles_dir, _models_dir) = + build_router_with_pool_limits(PoolLimits::for_pool_writes(true)); + let (status, pool) = post_pool(&app, never).await; + assert_eq!(status, StatusCode::CREATED); + assert_eq!(pool["spec"]["ttl_seconds"], "never"); + } + + #[tokio::test] + async fn test_create_pool_rejects_a_ttl_that_is_neither_seconds_nor_never() { + let (app, _bundles_dir, _models_dir) = + build_router_with_pool_limits(PoolLimits::for_pool_writes(true)); + for ttl in [serde_json::json!("forever"), serde_json::json!(-1)] { + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/pools") + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({"name": "rag", "gpus": {"l4": 1}, "ttl_seconds": ttl}) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert!(response.status().is_client_error(), "{ttl}"); + } + } + #[tokio::test] async fn test_files_routes_return_flat_404() { let (app, _bundles_dir, _models_dir) = build_router(); diff --git a/packages/sie_gateway/src/state/k8s_pool_backend.rs b/packages/sie_gateway/src/state/k8s_pool_backend.rs index 960b96c8c..ef530f0ea 100644 --- a/packages/sie_gateway/src/state/k8s_pool_backend.rs +++ b/packages/sie_gateway/src/state/k8s_pool_backend.rs @@ -503,7 +503,7 @@ mod tests { m }, gpu_caps: HashMap::new(), - ttl_seconds: Some(600), + ttl_seconds: Some(crate::types::pool::PoolTtl::Seconds(600)), minimum_worker_count: 2, pinned_models: Vec::new(), }, @@ -596,7 +596,7 @@ mod tests { m }, gpu_caps: HashMap::new(), - ttl_seconds: Some(300), + ttl_seconds: Some(crate::types::pool::PoolTtl::Seconds(300)), minimum_worker_count: 1, pinned_models: Vec::new(), }, diff --git a/packages/sie_gateway/src/state/pool_manager.rs b/packages/sie_gateway/src/state/pool_manager.rs index 388ddb616..c82d5beec 100644 --- a/packages/sie_gateway/src/state/pool_manager.rs +++ b/packages/sie_gateway/src/state/pool_manager.rs @@ -6,13 +6,19 @@ use tokio::sync::RwLock; use tracing::{info, warn}; use crate::state::k8s_pool_backend::K8sPoolBackend; -use crate::types::pool::{AssignedWorker, Pool, PoolSpec, PoolState, PoolStatus}; +use crate::types::pool::{AssignedWorker, Pool, PoolSpec, PoolState, PoolStatus, PoolTtl}; pub const DEFAULT_POOL_NAME: &str = "default"; -const DEFAULT_LEASE_DURATION_S: f64 = 1200.0; // 20 minutes +/// Lease of an API pool created without `ttl_seconds`. SDK clients renew it +/// every 60 s while they run, so it is the grace period after a client stops. +const DEFAULT_LEASE_SECONDS: u64 = 1200; const TIMESTAMP_TOLERANCE_S: f64 = 0.001; pub const DEFAULT_POOL_MAX_MINIMUM_WORKER_COUNT: u32 = 4; -pub const DEFAULT_POOL_MAX_TTL_SECONDS: u64 = 3600; +/// Default TTL limit when every pool write carries the admin token: 30 days. +pub const DEFAULT_POOL_MAX_TTL_SECONDS: u64 = 30 * 24 * 60 * 60; +/// Default TTL limit when pool writes are unauthenticated: one request must +/// not hold warm capacity for long. +pub const UNAUTHENTICATED_POOL_MAX_TTL_SECONDS: u64 = 60 * 60; pub const DEFAULT_MAX_POOLS: usize = 64; type WorkerAssignment = (String, String, String, String, String); @@ -24,28 +30,53 @@ pub struct PoolLimits { /// floor summed over the pool's machine profiles (`minimum_worker_count` /// applies to each profile) and the `gpus` requirement summed over /// profiles, and caps what a stored pool exports through its warm floor - /// and active lease. + /// and active lease. It bounds only the always-on floor: KEDA scales each + /// lane above it on demand, up to the lane's `maxReplicas`. pub max_minimum_worker_count: u32, - /// Largest accepted `ttl_seconds`; also caps the effective lease of any - /// stored pool, including pools restored from Kubernetes. + /// Largest accepted numeric `ttl_seconds`; also caps the effective lease + /// of any stored pool with a lease, including pools restored from + /// Kubernetes. Defaults to 30 days when pool writes are admin-gated and + /// to one hour when they are not. pub max_ttl_seconds: u64, /// Largest number of live API-created pools, checked by the replica that /// serves the create against every pool it knows, including pools applied /// from other replicas. Concurrent creates on different replicas can pass /// the check before either sees the other's pool. pub max_pools: usize, + /// Whether the API may create a pool with `ttl_seconds: "never"`. The + /// gateway allows it only with gateway auth on, where every pool write + /// needs the admin token; with auth off, one unauthenticated request + /// would otherwise hold warm capacity until someone deletes the pool. + pub allow_no_expiry: bool, } -impl Default for PoolLimits { - fn default() -> Self { +impl PoolLimits { + /// Default bounds for a gateway on which every pool write is checked + /// against the admin token (`admin_gated`), or on which it is not. Pass + /// `true` only when no caller without the admin token can reach + /// `POST /v1/pools`: it allows pools that never expire and leases of up + /// to 30 days, where an unauthenticated gateway keeps the one-hour limit. + pub fn for_pool_writes(admin_gated: bool) -> Self { Self { max_minimum_worker_count: DEFAULT_POOL_MAX_MINIMUM_WORKER_COUNT, - max_ttl_seconds: DEFAULT_POOL_MAX_TTL_SECONDS, + max_ttl_seconds: if admin_gated { + DEFAULT_POOL_MAX_TTL_SECONDS + } else { + UNAUTHENTICATED_POOL_MAX_TTL_SECONDS + }, max_pools: DEFAULT_MAX_POOLS, + allow_no_expiry: admin_gated, } } } +/// The unauthenticated bounds: the safe choice when nothing says otherwise. +impl Default for PoolLimits { + fn default() -> Self { + Self::for_pool_writes(false) + } +} + #[derive(Debug, PartialEq, Eq)] pub enum PoolLimitError { MinimumWorkerCount { @@ -65,6 +96,10 @@ pub enum PoolLimitError { requested: u64, max: u64, }, + ZeroTtl, + NoExpiryRequiresAuth { + max: u64, + }, TooManyPools { max: usize, }, @@ -93,6 +128,14 @@ impl std::fmt::Display for PoolLimitError { f, "ttl_seconds {requested} exceeds the gateway limit of {max} (SIE_GATEWAY_POOL_MAX_TTL_S)" ), + PoolLimitError::ZeroTtl => write!( + f, + "ttl_seconds must be at least 1; send \"never\" for a pool that never expires" + ), + PoolLimitError::NoExpiryRequiresAuth { max } => write!( + f, + "ttl_seconds \"never\" requires gateway auth (SIE_AUTH_MODE=token), so that only admin-token holders create pools that never expire; send a ttl_seconds of at most {max} instead" + ), PoolLimitError::TooManyPools { max } => write!( f, "Pool limit reached: the gateway already holds {max} API-created pools (SIE_GATEWAY_MAX_POOLS); delete a pool or let one expire first" @@ -242,7 +285,6 @@ impl CapacityPoolSnapshot { pub struct PoolManager { pools: RwLock>, - lease_duration_s: f64, configured_profiles: Vec, static_pool_names: RwLock>, /// Optional K8s ConfigMap backend for pool persistence. @@ -254,7 +296,6 @@ impl PoolManager { pub fn new(configured_profiles: Vec) -> Self { Self { pools: RwLock::new(HashMap::new()), - lease_duration_s: DEFAULT_LEASE_DURATION_S, configured_profiles, static_pool_names: RwLock::new(HashSet::new()), k8s_backend: None, @@ -330,17 +371,39 @@ impl PoolManager { .as_secs_f64() } - /// Lease TTL in whole seconds: the pool's own TTL or the gateway default, - /// capped at `PoolLimits::max_ttl_seconds`. - fn effective_ttl_seconds(&self, pool: &Pool) -> u64 { - pool.spec - .ttl_seconds - .unwrap_or(self.lease_duration_s as u64) - .min(self.limits.max_ttl_seconds) - } - - fn lease_ttl_seconds(&self, pool: &Pool) -> i32 { - self.effective_ttl_seconds(pool).min(i32::MAX as u64) as i32 + /// The one rule for how long a pool lives: seconds after its last renewal, + /// or `None` for a pool without a lease. The `default` pool, static Helm + /// queue pools, and API pools created with `ttl_seconds: "never"` all + /// carry `PoolTtl::Never`, so expiry, renewal, and Kubernetes Leases treat + /// them alike. A finite lease is the pool's own TTL or the gateway + /// default, capped at `PoolLimits::max_ttl_seconds`, including for pools + /// restored from Kubernetes or applied from another replica. + fn lease_seconds(&self, pool: &Pool) -> Option { + let ttl = match pool.spec.ttl_seconds { + Some(PoolTtl::Never) => return None, + Some(PoolTtl::Seconds(ttl)) => ttl, + None => DEFAULT_LEASE_SECONDS, + }; + Some(ttl.min(self.limits.max_ttl_seconds)) + } + + /// Make a stored pool's Kubernetes Lease match its lifetime: create or + /// renew it for a pool with a lease, and drop any left over for a pool + /// that never expires. Best-effort, like every backend write here. + async fn sync_k8s_lease(&self, backend: &K8sPoolBackend, key: &str, pool: &Pool) { + match self.lease_seconds(pool) { + Some(ttl) => { + let ttl = i32::try_from(ttl).unwrap_or(i32::MAX); + if let Err(e) = backend.create_or_renew_lease(key, ttl).await { + warn!(error = %e, pool = %key, "failed to create or renew K8s Lease"); + } + } + None => { + if let Err(e) = backend.delete_lease(key).await { + warn!(error = %e, pool = %key, "failed to delete K8s Lease of a pool that never expires"); + } + } + } } fn normalize_gpus_and_caps( @@ -457,7 +520,7 @@ impl PoolManager { fn check_pool_spec_limits( &self, gpus: &HashMap, - ttl_seconds: Option, + ttl_seconds: Option, minimum_worker_count: u32, ) -> Result<(), PoolLimitError> { let max = self.limits.max_minimum_worker_count; @@ -486,15 +549,18 @@ impl PoolManager { max, }); } - if let Some(ttl) = ttl_seconds { - if ttl > self.limits.max_ttl_seconds { - return Err(PoolLimitError::Ttl { - requested: ttl, - max: self.limits.max_ttl_seconds, - }); - } + let max_ttl = self.limits.max_ttl_seconds; + match ttl_seconds { + None => Ok(()), + Some(PoolTtl::Seconds(0)) => Err(PoolLimitError::ZeroTtl), + Some(PoolTtl::Seconds(ttl)) if ttl > max_ttl => Err(PoolLimitError::Ttl { + requested: ttl, + max: max_ttl, + }), + Some(PoolTtl::Seconds(_)) => Ok(()), + Some(PoolTtl::Never) if self.limits.allow_no_expiry => Ok(()), + Some(PoolTtl::Never) => Err(PoolLimitError::NoExpiryRequiresAuth { max: max_ttl }), } - Ok(()) } async fn is_static_pool(&self, name: &str) -> bool { @@ -538,7 +604,7 @@ impl PoolManager { bundle, gpus, gpu_caps: spec.gpu_caps.clone(), - ttl_seconds: None, + ttl_seconds: Some(PoolTtl::Never), minimum_worker_count: spec.minimum_worker_count, pinned_models: spec.pinned_models.clone(), }; @@ -643,7 +709,14 @@ impl PoolManager { .collect(); match self - .create_pool(DEFAULT_POOL_NAME, gpus, None, None, 0, Vec::new()) + .create_pool( + DEFAULT_POOL_NAME, + gpus, + None, + Some(PoolTtl::Never), + 0, + Vec::new(), + ) .await { Ok(_) => { @@ -663,7 +736,7 @@ impl PoolManager { name: &str, gpus: HashMap, bundle: Option, - ttl_seconds: Option, + ttl_seconds: Option, minimum_worker_count: u32, pinned_models: Vec, ) -> Result> { @@ -686,7 +759,7 @@ impl PoolManager { gpus: HashMap, gpu_caps: HashMap, bundle: Option, - ttl_seconds: Option, + ttl_seconds: Option, minimum_worker_count: u32, pinned_models: Vec, ) -> Result> { @@ -711,7 +784,7 @@ impl PoolManager { gpus: HashMap, gpu_caps: HashMap, bundle: Option, - ttl_seconds: Option, + ttl_seconds: Option, minimum_worker_count: u32, pinned_models: Vec, ) -> Result> { @@ -751,25 +824,31 @@ impl PoolManager { let existing = pools .get_mut(&pool_key) .expect("pool key resolved from the same map"); - let spec_changed = existing.spec.bundle != bundle + // An update that omits `ttl_seconds` keeps the pool's lifetime, so + // a client that re-posts a pool at startup never shortens one an + // admin made permanent. Send a number to give it a lease again. + let ttl_seconds = ttl_seconds.or(existing.spec.ttl_seconds); + let assignment_changed = existing.spec.bundle != bundle || existing.spec.queue_pool != queue_pool || existing.spec.gpus != gpus || existing.spec.gpu_caps != gpu_caps - || existing.spec.ttl_seconds != ttl_seconds || existing.spec.minimum_worker_count != minimum_worker_count || existing.spec.pinned_models != pinned_models; + // A lifetime change keeps the pool's workers: it changes when the + // pool expires, not what it serves. + let spec_changed = assignment_changed || existing.spec.ttl_seconds != ttl_seconds; if pool_key.eq_ignore_ascii_case(DEFAULT_POOL_NAME) && spec_changed { return Err(Box::new(DefaultPoolMutationError)); } existing.status.last_renewed = now; - if spec_changed { + existing.spec.ttl_seconds = ttl_seconds; + if assignment_changed { existing.spec.bundle = bundle; existing.spec.queue_pool = queue_pool.clone(); existing.spec.gpus = gpus; existing.spec.gpu_caps = gpu_caps; - existing.spec.ttl_seconds = ttl_seconds; existing.spec.minimum_worker_count = minimum_worker_count; existing.spec.pinned_models = pinned_models.clone(); existing.status.state = PoolState::Pending; @@ -787,10 +866,7 @@ impl PoolManager { warn!(error = %e, pool = %pool_key, "failed to persist pool update to K8s"); } } - let ttl = self.lease_ttl_seconds(&result); - if let Err(e) = backend.create_or_renew_lease(&pool_key, ttl).await { - warn!(error = %e, pool = %pool_key, "failed to renew K8s Lease"); - } + self.sync_k8s_lease(backend, &pool_key, &result).await; } } @@ -833,16 +909,13 @@ impl PoolManager { info!(pool = %name, "created pool"); // Persist dynamic named pools to K8s. The default pool is synthetic - // per gateway replica and is not a lease-owned runtime pool. + // per gateway replica and is not stored. if !name.eq_ignore_ascii_case(DEFAULT_POOL_NAME) { if let Some(ref backend) = self.k8s_backend { if let Err(e) = backend.save_pool(&pool).await { warn!(error = %e, pool = %name, "failed to persist pool to K8s"); } - let ttl = self.lease_ttl_seconds(&pool); - if let Err(e) = backend.create_or_renew_lease(&name, ttl).await { - warn!(error = %e, pool = %name, "failed to create K8s Lease"); - } + self.sync_k8s_lease(backend, &name, &pool).await; } } @@ -1025,34 +1098,33 @@ impl PoolManager { Ok(removed) } + /// Renew a pool's lease. Renewing a pool that never expires succeeds and + /// changes nothing that matters. pub async fn renew_pool(&self, name: &str) -> bool { - let is_static_pool = self.is_static_pool(name).await; - let found_key = { + let (key, has_lease) = { let mut pools = self.pools.write().await; - if let Some(key) = pool_key_for_name(&pools, name) { - if let Some(pool) = pools.get_mut(&key) { - pool.status.last_renewed = Self::now_secs(); - Some(key) - } else { - None - } - } else { - None - } + let Some(key) = pool_key_for_name(&pools, name) else { + return false; + }; + let pool = pools + .get_mut(&key) + .expect("pool key resolved from the same map"); + pool.status.last_renewed = Self::now_secs(); + let has_lease = self.lease_seconds(pool).is_some(); + (key, has_lease) }; // write lock dropped here - // Renew the K8s Lease (best-effort, skip default and static Helm pools) - if let Some(key) = found_key.as_ref() { + // Renew the K8s Lease (best-effort). Pools that never expire, which + // include the default and static Helm pools, have none. + if has_lease { if let Some(ref backend) = self.k8s_backend { - if !key.eq_ignore_ascii_case(DEFAULT_POOL_NAME) && !is_static_pool { - if let Err(e) = backend.renew_lease(key).await { - warn!(error = %e, pool = %key, "failed to renew K8s Lease"); - } + if let Err(e) = backend.renew_lease(&key).await { + warn!(error = %e, pool = %key, "failed to renew K8s Lease"); } } } - found_key.is_some() + true } pub async fn assign_workers( @@ -1279,15 +1351,14 @@ impl PoolManager { let static_pool_names = self.static_pool_names.read().await.clone(); { + // Pools without a lease (the default pool, static Helm queue + // pools, and API pools that never expire) are never collected. let pools = self.pools.read().await; for (name, pool) in pools.iter() { - if name.eq_ignore_ascii_case(DEFAULT_POOL_NAME) - || static_pool_names.contains(&normalize_pool_name(name)) - { + let Some(ttl) = self.lease_seconds(pool) else { continue; - } - let ttl = self.effective_ttl_seconds(pool) as f64; - if now - pool.status.last_renewed > ttl { + }; + if now - pool.status.last_renewed > ttl as f64 { expired.push(name.clone()); } } @@ -1305,11 +1376,34 @@ impl PoolManager { { continue; } - let pools = self.pools.read().await; - let in_memory = pools.contains_key(&name); - drop(pools); - - if in_memory { + let in_memory_lease = { + let pools = self.pools.read().await; + pools.get(&name).map(|pool| self.lease_seconds(pool)) + }; + // A Lease left over from before a pool stopped + // expiring must not collect it. A pool this replica + // does not know is read back from its ConfigMap + // first, so it is not mistaken for an orphan; one it + // cannot read is left for a later sweep, never deleted. + let never_expires = match in_memory_lease { + Some(lease) => lease.is_none(), + None => match backend.load_pool(&name).await { + Ok(stored) => stored.is_some_and(|stored| { + stored.spec.ttl_seconds == Some(PoolTtl::Never) + }), + Err(e) => { + warn!(pool = %name, error = %e, "cannot read the pool behind an expired K8s Lease; leaving both for a later sweep"); + continue; + } + }, + }; + + if never_expires { + info!(pool = %name, "dropping stale K8s Lease of a pool that never expires"); + if let Err(e) = backend.delete_lease(&name).await { + warn!(pool = %name, error = %e, "failed to delete stale Lease from K8s"); + } + } else if in_memory_lease.is_some() { info!(pool = %name, "K8s Lease expired (crash-safe TTL)"); expired.push(name); } else { @@ -1629,13 +1723,21 @@ mod tests { let mut gpu_caps = HashMap::new(); gpu_caps.insert("l4-spot".to_string(), 2); let pool = pm - .create_pool_with_caps("test", updated_gpus, gpu_caps, None, Some(60), 0, vec![]) + .create_pool_with_caps( + "test", + updated_gpus, + gpu_caps, + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) .await .unwrap(); assert_eq!(pool.spec.gpus.get("l4-spot"), Some(&0)); assert_eq!(pool.spec.gpu_caps.get("l4-spot"), Some(&2)); - assert_eq!(pool.spec.ttl_seconds, Some(60)); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Seconds(60))); assert_eq!(pool.status.state, PoolState::Pending); assert!(pool.status.assigned_workers.is_empty()); } @@ -1704,13 +1806,20 @@ mod tests { let mut updated_gpus = HashMap::new(); updated_gpus.insert("l4-spot".to_string(), 2); let pool = pm - .create_pool("bench", updated_gpus, None, Some(60), 0, vec![]) + .create_pool( + "bench", + updated_gpus, + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) .await .unwrap(); assert_eq!(pool.spec.name, "bench"); assert_eq!(pool.spec.gpus.get("l4-spot"), Some(&2)); - assert_eq!(pool.spec.ttl_seconds, Some(60)); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Seconds(60))); let pools = pm.list_pools().await; assert_eq!(pools.len(), 1); } @@ -1756,7 +1865,7 @@ mod tests { bundle: None, gpus, gpu_caps: HashMap::new(), - ttl_seconds: Some(1), + ttl_seconds: Some(PoolTtl::Seconds(1)), minimum_worker_count: 0, pinned_models: Vec::new(), }]; @@ -1765,7 +1874,7 @@ mod tests { assert_eq!(count, 1); let pool = pm.get_pool("company-a").await.unwrap(); - assert_eq!(pool.spec.ttl_seconds, None); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Never)); assert!(pool.spec.gpu_caps.is_empty()); assert!(pm .admitted_worker_names_for_capped_lane("company-a", "l4-spot", "default") @@ -1775,7 +1884,7 @@ mod tests { { let mut pools = pm.pools.write().await; pools.get_mut("company-a").unwrap().status.last_renewed -= - DEFAULT_LEASE_DURATION_S + 10.0; + (DEFAULT_POOL_MAX_TTL_SECONDS + 10) as f64; } let expired = pm.check_expired_leases().await; @@ -1839,7 +1948,14 @@ mod tests { let mut changed_gpus = HashMap::new(); changed_gpus.insert("l4-spot".to_string(), 1); let result = pm - .create_pool("bench", changed_gpus, None, Some(60), 0, vec![]) + .create_pool( + "bench", + changed_gpus, + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) .await; assert!(result.is_err()); assert!(result @@ -1879,7 +1995,14 @@ mod tests { let mut changed_gpus = HashMap::new(); changed_gpus.insert("l4-spot".to_string(), 1); let result = pm - .create_pool("BENCH", changed_gpus, None, Some(60), 0, vec![]) + .create_pool( + "BENCH", + changed_gpus, + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) .await; assert!(result.is_err()); assert!(result @@ -1964,9 +2087,16 @@ mod tests { let mut dynamic_gpus = HashMap::new(); dynamic_gpus.insert("l4-spot".to_string(), 1); - pm.create_pool("Bench", dynamic_gpus, None, Some(60), 0, vec![]) - .await - .unwrap(); + pm.create_pool( + "Bench", + dynamic_gpus, + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) + .await + .unwrap(); let mut static_gpus = HashMap::new(); static_gpus.insert("l4-spot".to_string(), 0); @@ -2016,7 +2146,7 @@ mod tests { bundle: Some("other".to_string()), gpus: remote_gpus, gpu_caps: HashMap::new(), - ttl_seconds: Some(60), + ttl_seconds: Some(PoolTtl::Seconds(60)), minimum_worker_count: 1, pinned_models: Vec::new(), }, @@ -2037,7 +2167,7 @@ mod tests { let found = pm.get_pool("bench").await.unwrap(); assert_eq!(found.spec.bundle, None); assert_eq!(found.spec.gpus.get("l4-spot"), Some(&0)); - assert_eq!(found.spec.ttl_seconds, None); + assert_eq!(found.spec.ttl_seconds, Some(PoolTtl::Never)); assert_eq!(found.status.state, PoolState::Pending); assert!(found.status.assigned_workers.is_empty()); @@ -2515,9 +2645,16 @@ mod tests { gpus.insert("l4-spot".to_string(), 1); // Pool with short TTL (1 second) - pm.create_pool("short-ttl", gpus.clone(), None, Some(1), 0, vec![]) - .await - .unwrap(); + pm.create_pool( + "short-ttl", + gpus.clone(), + None, + Some(PoolTtl::Seconds(1)), + 0, + vec![], + ) + .await + .unwrap(); // Pool with no TTL (uses global default = 1200s) pm.create_pool("default-ttl", gpus, None, None, 0, vec![]) @@ -2693,7 +2830,14 @@ mod tests { let mut gpus = HashMap::new(); gpus.insert("l4-spot".to_string(), 0); let local = pm - .create_pool("bench", gpus.clone(), None, Some(60), 0, vec![]) + .create_pool( + "bench", + gpus.clone(), + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) .await .unwrap(); @@ -2725,7 +2869,7 @@ mod tests { "bench", gpus, Some("default".to_string()), - Some(60), + Some(PoolTtl::Seconds(60)), 0, vec![], ) @@ -2754,7 +2898,7 @@ mod tests { "bench", gpus, Some("default".to_string()), - Some(60), + Some(PoolTtl::Seconds(60)), 0, vec![], ) @@ -2780,13 +2924,13 @@ mod tests { let mut gpus = HashMap::new(); gpus.insert("l4-spot".to_string(), 1); let local = pm - .create_pool("Bench", gpus, None, Some(60), 0, vec![]) + .create_pool("Bench", gpus, None, Some(PoolTtl::Seconds(60)), 0, vec![]) .await .unwrap(); let mut remote = local.clone(); remote.spec.name = "bench".to_string(); - remote.spec.ttl_seconds = Some(120); + remote.spec.ttl_seconds = Some(PoolTtl::Seconds(120)); remote.status.last_renewed += 1.0; pm.apply_remote_pool(remote).await; @@ -2795,7 +2939,7 @@ mod tests { assert_eq!(pools.len(), 1); let found = pm.get_pool("Bench").await.unwrap(); assert_eq!(found.spec.name, "bench"); - assert_eq!(found.spec.ttl_seconds, Some(120)); + assert_eq!(found.spec.ttl_seconds, Some(PoolTtl::Seconds(120))); } #[tokio::test] @@ -2809,7 +2953,7 @@ mod tests { bundle: None, gpus: HashMap::from([("l4-spot".to_string(), 1)]), gpu_caps: HashMap::new(), - ttl_seconds: Some(60), + ttl_seconds: Some(PoolTtl::Seconds(60)), minimum_worker_count: 0, pinned_models: Vec::new(), }, @@ -2842,7 +2986,7 @@ mod tests { let mut gpus = HashMap::new(); gpus.insert("l4-spot".to_string(), 1); - pm.create_pool("Bench", gpus, None, Some(60), 0, vec![]) + pm.create_pool("Bench", gpus, None, Some(PoolTtl::Seconds(60)), 0, vec![]) .await .unwrap(); @@ -2874,7 +3018,7 @@ mod tests { } } - fn remote_pool(name: &str, ttl_seconds: Option, minimum_worker_count: u32) -> Pool { + fn remote_pool(name: &str, ttl_seconds: Option, minimum_worker_count: u32) -> Pool { Pool { spec: PoolSpec { name: name.to_string(), @@ -2900,6 +3044,13 @@ mod tests { let limits = PoolLimits::default(); assert_eq!(limits.max_minimum_worker_count, 4); assert_eq!(limits.max_ttl_seconds, 3600); + assert!(!limits.allow_no_expiry); + + let admin_gated = PoolLimits::for_pool_writes(true); + assert_eq!(admin_gated.max_minimum_worker_count, 4); + assert_eq!(admin_gated.max_ttl_seconds, 30 * 24 * 3600); + assert_eq!(admin_gated.max_pools, 64); + assert!(admin_gated.allow_no_expiry); assert_eq!(limits.max_pools, 64); } @@ -3205,11 +3356,25 @@ mod tests { ..PoolLimits::default() }); - pm.create_pool("at-limit", l4_gpus(), None, Some(600), 0, vec![]) - .await - .expect("a TTL at the limit is accepted"); + pm.create_pool( + "at-limit", + l4_gpus(), + None, + Some(PoolTtl::Seconds(600)), + 0, + vec![], + ) + .await + .expect("a TTL at the limit is accepted"); let error = pm - .create_pool("above-limit", l4_gpus(), None, Some(u64::MAX), 0, vec![]) + .create_pool( + "above-limit", + l4_gpus(), + None, + Some(PoolTtl::Seconds(u64::MAX)), + 0, + vec![], + ) .await .expect_err("a TTL above the limit is rejected"); @@ -3245,9 +3410,16 @@ mod tests { .expect_err("a pool beyond the limit is rejected"); assert_eq!(limit_error(error), PoolLimitError::TooManyPools { max: 2 }); - pm.create_pool("SECOND", l4_gpus(), None, Some(60), 0, vec![]) - .await - .expect("updating an existing pool at the limit is allowed"); + pm.create_pool( + "SECOND", + l4_gpus(), + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) + .await + .expect("updating an existing pool at the limit is allowed"); pm.delete_pool("first").await.unwrap(); pm.create_pool("third", l4_gpus(), None, None, 0, vec![]) @@ -3277,8 +3449,12 @@ mod tests { max_ttl_seconds: 60, ..PoolLimits::default() }); - pm.apply_remote_pool(remote_pool("legacy", Some(u64::from(u32::MAX)), 0)) - .await; + pm.apply_remote_pool(remote_pool( + "legacy", + Some(PoolTtl::Seconds(u64::from(u32::MAX))), + 0, + )) + .await; { let mut pools = pm.pools.write().await; pools.get_mut("legacy").unwrap().status.last_renewed -= 61.0; @@ -3300,7 +3476,7 @@ mod tests { .await .unwrap(); let pool = pm.get_pool("no-ttl").await.unwrap(); - assert_eq!(pm.lease_ttl_seconds(&pool), 600); + assert_eq!(pm.lease_seconds(&pool), Some(600)); { let mut pools = pm.pools.write().await; pools.get_mut("no-ttl").unwrap().status.last_renewed -= 601.0; @@ -3317,7 +3493,236 @@ mod tests { .unwrap(); let pool = pm.get_pool("no-ttl").await.unwrap(); - assert_eq!(pm.lease_ttl_seconds(&pool), DEFAULT_LEASE_DURATION_S as i32); + assert_eq!(pm.lease_seconds(&pool), Some(DEFAULT_LEASE_SECONDS)); + } + + fn no_expiry_pool_manager() -> PoolManager { + limited_pool_manager(PoolLimits::for_pool_writes(true)) + } + + async fn age_pool(pm: &PoolManager, name: &str, seconds: u64) { + let mut pools = pm.pools.write().await; + pools.get_mut(name).unwrap().status.last_renewed -= seconds as f64; + } + + #[tokio::test] + async fn test_pool_that_never_expires_outlives_every_lease() { + let pm = no_expiry_pool_manager(); + pm.create_pool("rag", l4_gpus(), None, Some(PoolTtl::Never), 1, vec![]) + .await + .expect("a pool that never expires is accepted when gateway auth is on"); + let pool = pm.get_pool("rag").await.unwrap(); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Never)); + assert_eq!(pm.lease_seconds(&pool), None); + + age_pool(&pm, "rag", 10 * DEFAULT_POOL_MAX_TTL_SECONDS).await; + + assert!(pm.check_expired_leases().await.is_empty()); + assert!( + pm.renew_pool("rag").await, + "renewing it is a harmless no-op" + ); + assert!(pm.get_pool("rag").await.is_some()); + } + + #[tokio::test] + async fn test_pool_that_never_expires_requires_gateway_auth() { + let pm = PoolManager::new(vec!["l4-spot".to_string()]); + let error = pm + .create_pool("rag", l4_gpus(), None, Some(PoolTtl::Never), 0, vec![]) + .await + .expect_err("without gateway auth a pool must expire"); + + assert_eq!( + limit_error(error), + PoolLimitError::NoExpiryRequiresAuth { + max: UNAUTHENTICATED_POOL_MAX_TTL_SECONDS + } + ); + assert!(pm.get_pool("rag").await.is_none()); + } + + #[tokio::test] + async fn test_pool_that_never_expires_keeps_the_warm_floor_budget() { + let pm = no_expiry_pool_manager(); + let error = pm + .create_pool("rag", l4_gpus(), None, Some(PoolTtl::Never), 5, vec![]) + .await + .expect_err("no expiry does not lift the warm-floor budget"); + + assert!(matches!( + limit_error(error), + PoolLimitError::MinimumWorkerCount { + requested: 5, + max: 4 + } + )); + } + + #[tokio::test] + async fn test_zero_ttl_is_rejected_rather_than_read_as_no_expiry() { + let pm = no_expiry_pool_manager(); + let error = pm + .create_pool( + "bench", + l4_gpus(), + None, + Some(PoolTtl::Seconds(0)), + 0, + vec![], + ) + .await + .expect_err("a zero TTL would expire the pool at the next sweep"); + + assert_eq!(limit_error(error), PoolLimitError::ZeroTtl); + } + + #[tokio::test] + async fn test_unauthenticated_pool_writes_keep_the_one_hour_ttl_limit() { + let pm = PoolManager::new(vec!["l4-spot".to_string()]); + let error = pm + .create_pool( + "eval", + l4_gpus(), + None, + Some(PoolTtl::Seconds(3601)), + 0, + vec![], + ) + .await + .expect_err("without gateway auth a lease stays within an hour"); + + assert_eq!( + limit_error(error), + PoolLimitError::Ttl { + requested: 3601, + max: 3600 + } + ); + } + + #[tokio::test] + async fn test_ttls_far_beyond_an_hour_are_accepted_with_gateway_auth() { + let pm = no_expiry_pool_manager(); + let week = 7 * 24 * 3600; + pm.create_pool( + "eval", + l4_gpus(), + None, + Some(PoolTtl::Seconds(week)), + 0, + vec![], + ) + .await + .unwrap(); + let pool = pm.get_pool("eval").await.unwrap(); + assert_eq!(pm.lease_seconds(&pool), Some(week)); + + age_pool(&pm, "eval", week - 60).await; + assert!(pm.check_expired_leases().await.is_empty()); + age_pool(&pm, "eval", 120).await; + assert_eq!(pm.check_expired_leases().await, vec!["eval".to_string()]); + } + + #[tokio::test] + async fn test_update_without_a_ttl_keeps_the_pools_lifetime() { + let pm = no_expiry_pool_manager(); + pm.create_pool("rag", l4_gpus(), None, Some(PoolTtl::Never), 1, vec![]) + .await + .unwrap(); + + // A service re-posting the pool at startup without a TTL must not + // shorten a pool an admin made permanent. + let pool = pm + .create_pool("rag", l4_gpus(), None, None, 1, vec![]) + .await + .unwrap(); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Never)); + + let pool = pm + .create_pool( + "rag", + l4_gpus(), + None, + Some(PoolTtl::Seconds(600)), + 1, + vec![], + ) + .await + .unwrap(); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Seconds(600))); + let pool = pm + .create_pool("rag", l4_gpus(), None, None, 1, vec![]) + .await + .unwrap(); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Seconds(600))); + } + + #[tokio::test] + async fn test_changing_only_the_ttl_keeps_the_pools_workers() { + let pm = no_expiry_pool_manager(); + pm.create_pool("rag", l4_gpus(), None, None, 0, vec![]) + .await + .unwrap(); + let workers = vec![worker("w1", "http://w1:8080", "l4-spot", "default")]; + assert!(pm.assign_workers("rag", &workers).await); + + let pool = pm + .create_pool("rag", l4_gpus(), None, Some(PoolTtl::Never), 0, vec![]) + .await + .unwrap(); + + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Never)); + assert_eq!(pool.status.state, PoolState::Active); + assert_eq!(pool.status.assigned_workers.len(), 1); + age_pool(&pm, "rag", 2 * DEFAULT_LEASE_SECONDS).await; + assert!(pm.check_expired_leases().await.is_empty()); + + // Back to a lease: the pool expires again once it lapses. + pm.create_pool( + "rag", + l4_gpus(), + None, + Some(PoolTtl::Seconds(60)), + 0, + vec![], + ) + .await + .unwrap(); + age_pool(&pm, "rag", 61).await; + assert_eq!(pm.check_expired_leases().await, vec!["rag".to_string()]); + } + + #[tokio::test] + async fn test_stored_pool_that_never_expires_is_honoured_on_every_replica() { + // An admin created it on a replica with gateway auth on. A replica + // that would refuse to create one still keeps it rather than + // collecting it: refusing is about who may create such a pool. + let pm = PoolManager::new(vec!["l4-spot".to_string()]); + pm.apply_remote_pool(remote_pool("rag", Some(PoolTtl::Never), 1)) + .await; + age_pool(&pm, "rag", 10 * DEFAULT_POOL_MAX_TTL_SECONDS).await; + + assert!(pm.check_expired_leases().await.is_empty()); + let snapshot = pm.capacity_pools().await.into_iter().next().unwrap(); + assert_eq!(snapshot.minimum_worker_count, 1); + } + + #[tokio::test] + async fn test_operator_pools_carry_no_expiry_in_their_spec() { + let pm = PoolManager::new(vec!["l4-spot".to_string()]); + pm.create_default_pool().await; + pm.sync_static_pools(&[static_queue_pool("company-a", "l4-spot")]) + .await + .unwrap(); + + for name in [DEFAULT_POOL_NAME, "company-a"] { + let pool = pm.get_pool(name).await.unwrap(); + assert_eq!(pool.spec.ttl_seconds, Some(PoolTtl::Never), "{name}"); + assert_eq!(pm.lease_seconds(&pool), None, "{name}"); + age_pool(&pm, name, 10 * DEFAULT_POOL_MAX_TTL_SECONDS).await; + } + assert!(pm.check_expired_leases().await.is_empty()); } fn assigned(count: usize, gpu: &str) -> Vec { diff --git a/packages/sie_gateway/src/types/pool.rs b/packages/sie_gateway/src/types/pool.rs index 36397ac0a..cd6ccc659 100644 --- a/packages/sie_gateway/src/types/pool.rs +++ b/packages/sie_gateway/src/types/pool.rs @@ -1,7 +1,93 @@ -use serde::{Deserialize, Serialize}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; use std::collections::HashMap; use utoipa::ToSchema; +/// The wire spelling of [`PoolTtl::Never`]. +pub const POOL_TTL_NEVER: &str = "never"; + +/// How long a pool lives after its last renewal. On the wire it is a whole +/// number of seconds or the string `"never"`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PoolTtl { + /// The pool expires this many seconds after its last renewal. + Seconds(u64), + /// The pool has no lease: it lives until it is deleted. + Never, +} + +impl Serialize for PoolTtl { + fn serialize(&self, serializer: S) -> Result { + match self { + PoolTtl::Seconds(seconds) => serializer.serialize_u64(*seconds), + PoolTtl::Never => serializer.serialize_str(POOL_TTL_NEVER), + } + } +} + +impl<'de> Deserialize<'de> for PoolTtl { + fn deserialize>(deserializer: D) -> Result { + struct PoolTtlVisitor; + + impl serde::de::Visitor<'_> for PoolTtlVisitor { + type Value = PoolTtl; + + fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(r#"a whole number of seconds or "never""#) + } + + fn visit_u64(self, seconds: u64) -> Result { + Ok(PoolTtl::Seconds(seconds)) + } + + fn visit_i64(self, seconds: i64) -> Result { + u64::try_from(seconds) + .map(PoolTtl::Seconds) + .map_err(|_| E::invalid_value(serde::de::Unexpected::Signed(seconds), &self)) + } + + fn visit_str(self, value: &str) -> Result { + if value == POOL_TTL_NEVER { + Ok(PoolTtl::Never) + } else { + Err(E::invalid_value(serde::de::Unexpected::Str(value), &self)) + } + } + } + + deserializer.deserialize_any(PoolTtlVisitor) + } +} + +impl utoipa::PartialSchema for PoolTtl { + fn schema() -> utoipa::openapi::RefOr { + use utoipa::openapi::schema::{ + KnownFormat, ObjectBuilder, OneOfBuilder, SchemaFormat, SchemaType, Type, + }; + OneOfBuilder::new() + .item( + ObjectBuilder::new() + .schema_type(SchemaType::Type(Type::Integer)) + .format(Some(SchemaFormat::KnownFormat(KnownFormat::Int64))) + .minimum(Some(1)) + .description(Some( + "The pool expires this many seconds after its last renewal.", + )), + ) + .item( + ObjectBuilder::new() + .schema_type(SchemaType::Type(Type::String)) + .enum_values(Some([POOL_TTL_NEVER])) + .description(Some("The pool never expires; it lives until it is deleted.")), + ) + .description(Some( + "How long a pool lives after its last renewal: a whole number of seconds, or \"never\".", + )) + .into() + } +} + +impl ToSchema for PoolTtl {} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] #[serde(rename_all = "lowercase")] pub enum PoolState { @@ -36,8 +122,12 @@ pub struct PoolSpec { pub gpus: HashMap, #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub gpu_caps: HashMap, + /// How long the pool lives after its last renewal: whole seconds, or + /// `"never"` for a pool that lives until it is deleted. `null` on an + /// API-created pool means the gateway's default lease. The `default` pool + /// and static Helm queue pools report `"never"`. #[serde(default)] - pub ttl_seconds: Option, + pub ttl_seconds: Option, /// Per-pool warm floor: the minimum number of machines the gateway keeps /// warm (via KEDA) so the first request to the pool never hits a cold VM. /// The gateway publishes it as `sie.gateway.pool.warm_floor`; the @@ -190,6 +280,57 @@ mod tests { assert_eq!(pool.worker_urls(), vec!["http://w1:8080", "http://w2:8080"]); } + #[test] + fn test_pool_ttl_wire_format() { + assert_eq!( + serde_json::to_value(PoolTtl::Seconds(86_400)).unwrap(), + serde_json::json!(86_400) + ); + assert_eq!( + serde_json::to_value(PoolTtl::Never).unwrap(), + serde_json::json!("never") + ); + assert_eq!( + serde_json::from_value::(serde_json::json!(60)).unwrap(), + PoolTtl::Seconds(60) + ); + assert_eq!( + serde_json::from_value::(serde_json::json!("never")).unwrap(), + PoolTtl::Never + ); + for invalid in [ + serde_json::json!(-1), + serde_json::json!(1.5), + serde_json::json!("forever"), + serde_json::json!("Never"), + serde_json::json!(true), + ] { + let error = serde_json::from_value::(invalid.clone()) + .expect_err("only whole seconds or \"never\" are a pool TTL"); + assert!( + error + .to_string() + .contains(r#"a whole number of seconds or "never""#), + "{invalid}: {error}" + ); + } + } + + #[test] + fn test_pool_spec_ttl_round_trips_through_stored_json() { + let mut pool = make_pool(PoolState::Active, vec![]); + for ttl in [None, Some(PoolTtl::Seconds(3600)), Some(PoolTtl::Never)] { + pool.spec.ttl_seconds = ttl; + let stored = serde_json::to_string(&pool).unwrap(); + let restored: Pool = serde_json::from_str(&stored).unwrap(); + assert_eq!(restored.spec.ttl_seconds, ttl, "{stored}"); + } + // Pools stored before `"never"` existed still parse. + let legacy = r#"{"spec":{"name":"legacy","gpus":{"l4":1},"ttl_seconds":600},"status":{"state":"active"}}"#; + let restored: Pool = serde_json::from_str(legacy).unwrap(); + assert_eq!(restored.spec.ttl_seconds, Some(PoolTtl::Seconds(600))); + } + #[test] fn test_pool_state_display() { assert_eq!(PoolState::Pending.to_string(), "pending"); diff --git a/packages/sie_sdk/src/sie_sdk/__init__.py b/packages/sie_sdk/src/sie_sdk/__init__.py index 535d8e217..9c6115244 100644 --- a/packages/sie_sdk/src/sie_sdk/__init__.py +++ b/packages/sie_sdk/src/sie_sdk/__init__.py @@ -92,6 +92,7 @@ PoolSpec, PoolSpecResponse, PoolStatusInfo, + PoolTtl, RateIdentity, ScoreResult, SparseResult, @@ -161,6 +162,7 @@ "PoolSpec", "PoolSpecResponse", "PoolStatusInfo", + "PoolTtl", "ProvisioningError", "RateIdentity", "RateLimitError", diff --git a/packages/sie_sdk/src/sie_sdk/client/_shared.py b/packages/sie_sdk/src/sie_sdk/client/_shared.py index 01d146014..61e4c498d 100644 --- a/packages/sie_sdk/src/sie_sdk/client/_shared.py +++ b/packages/sie_sdk/src/sie_sdk/client/_shared.py @@ -460,7 +460,27 @@ def attempt_timeout_cap(remaining: float) -> float: # Pool settings -DEFAULT_LEASE_RENEWAL_INTERVAL_S = 60.0 # Renew lease every 60s (lease is 1200s) +DEFAULT_LEASE_RENEWAL_INTERVAL_S = 60.0 # Renew lease every 60s (default lease is 1200s) + +#: ``ttl_seconds`` for a pool that lives until it is deleted. Such a pool has no +#: lease, so clients do not renew it. +POOL_TTL_NEVER = "never" + + +def validate_pool_ttl(ttl_seconds: object) -> None: + """Refuse a ``ttl_seconds`` the gateway would reject, before any request.""" + if ttl_seconds is None or ttl_seconds == POOL_TTL_NEVER: + return + if isinstance(ttl_seconds, bool) or not isinstance(ttl_seconds, int) or ttl_seconds < 1: + msg = f'ttl_seconds must be a positive whole number of seconds or "never", got {ttl_seconds!r}' + raise ValueError(msg) + + +def pool_never_expires(body: Any) -> bool: + """Whether a gateway pool response describes a pool without a lease.""" + spec = body.get("spec") if isinstance(body, dict) else None + return isinstance(spec, dict) and spec.get("ttl_seconds") == POOL_TTL_NEVER + # LoRA loading retry settings LORA_LOADING_MAX_RETRIES = 10 # Max retries for LoRA loading (usually completes in 1-2s) diff --git a/packages/sie_sdk/src/sie_sdk/client/async_.py b/packages/sie_sdk/src/sie_sdk/client/async_.py index 9f9698a98..993ea811f 100644 --- a/packages/sie_sdk/src/sie_sdk/client/async_.py +++ b/packages/sie_sdk/src/sie_sdk/client/async_.py @@ -87,6 +87,7 @@ OutputType, PoolInfo, PoolSpec, + PoolTtl, Recommendation, RequestMetadata, ResponseInputMessage, @@ -156,6 +157,7 @@ parse_score_result, parse_terminal_json_object, parse_terminal_msgpack_object, + pool_never_expires, provisioning_retry_delay, raise_if_estimate_unroutable, raise_if_input_too_long, @@ -175,6 +177,7 @@ validate_encode_traceparent, validate_generate_grammar, validate_generate_request_body, + validate_pool_ttl, websocket_matches_base_url_origin, ) from ._sse import SSECompletion, aiter_sse_payloads @@ -864,6 +867,8 @@ async def _ensure_pool_created(self) -> None: request_body["minimum_worker_count"] = self._pool_spec["minimum_worker_count"] if self._pool_spec.get("pinned_models") is not None: request_body["pinned_models"] = self._pool_spec["pinned_models"] + if self._pool_spec.get("ttl_seconds") is not None: + request_body["ttl_seconds"] = self._pool_spec["ttl_seconds"] try: response = await self._post( @@ -891,8 +896,9 @@ async def _ensure_pool_created(self) -> None: self._pool_created = True - # Start lease renewal task - await self._start_lease_renewal() + # A pool that never expires has no lease to renew. + if not pool_never_expires(data): + await self._start_lease_renewal() async def _start_lease_renewal(self) -> None: """Start the async lease renewal task.""" @@ -1000,6 +1006,7 @@ async def create_pool( pinned_models: list[str] | None = None, *, queue_pool: str | None = None, + ttl_seconds: PoolTtl | None = None, ) -> None: """Create or update a resource pool for isolated capacity. @@ -1012,24 +1019,38 @@ async def create_pool( {"l4": 4}. If omitted, all matching workers can be assigned. bundle: Optional bundle filter. When set, only workers running this bundle will be assigned to the pool. - minimum_worker_count: Per-pool warm floor (minimum machines kept warm). - The gateway emits canonical ``sie.gateway.pool.warm_floor`` - telemetry; the collector exposes ``sie_gateway_pool_warm_floor`` - to KEDA. Defaults to 0 (scale to zero). + minimum_worker_count: Per-pool warm floor: machines kept warm even with + no traffic. It is only the always-on floor: the lane autoscales + above it on demand, up to its KEDA maximum. The gateway emits + canonical ``sie.gateway.pool.warm_floor`` telemetry; the collector + exposes ``sie_gateway_pool_warm_floor`` to KEDA. Defaults to 0 + (scale to zero). pinned_models: Optional set of model ids to keep loaded so the first request to them pays no cold model-load. Each id must be a model the gateway already tracks and may be profile-qualified (``model-name:profile_name``); unknown ids are rejected. Defaults to none. queue_pool: Optional Helm/NATS queue namespace backing this logical pool. Defaults to "default", drawing from base capacity. + ttl_seconds: How long the pool lives after this client stops renewing + it. Omit for the gateway default (1200 s) on a new pool, or to keep + an existing pool's lifetime. While the client runs it renews the + pool every 60 s, so the pool lasts as long as the client plus this + grace period. Whole seconds, up to the gateway limit + (``SIE_GATEWAY_POOL_MAX_TTL_S``: by default 30 days with gateway + auth on, 1 hour without). ``"never"`` creates a pool that lives + until it is deleted, so it outlives this process and the client + does not renew it; it needs gateway auth (an admin ``api_key``). Raises: PoolError: If pool creation fails (e.g., invalid machine profile). SIEConnectionError: If unable to connect to the server. + ValueError: If ``minimum_worker_count`` is negative or ``ttl_seconds`` + is neither a positive whole number nor ``"never"``. """ if minimum_worker_count is not None and minimum_worker_count < 0: msg = "minimum_worker_count must be >= 0" raise ValueError(msg) + validate_pool_ttl(ttl_seconds) creation: _PoolCreationInFlight | None = None wait_for_creation: asyncio.Future[None] | None = None @@ -1074,6 +1095,8 @@ async def create_pool( request_body["minimum_worker_count"] = minimum_worker_count if pinned_models is not None: request_body["pinned_models"] = pinned_models + if ttl_seconds is not None: + request_body["ttl_seconds"] = ttl_seconds try: response = await self._post( @@ -1117,9 +1140,16 @@ async def create_pool( state = data.get("status", {}).get("state", "unknown") logger.info("Pool '%s' created/updated with state '%s'", name, state) + # A pool that never expires has no lease: release this call's + # creation slot, or stop the renewal task of a pool this client was + # renewing, instead of starting one. + if pool_never_expires(data): + await self._stop_pool_lease_renewal(name, creation) + return + # Start lease renewal task for this pool if this client is not # already tracking it. Repeated create_pool calls intentionally still - # POST so callers can update gpus/gpu_caps on the gateway. + # POST so callers can update the pool on the gateway. if reserved_slot and creation is not None: try: await self._start_pool_lease_renewal(name, creation) @@ -1131,6 +1161,25 @@ async def create_pool( raise _complete_pool_creation(creation.future) + async def _stop_pool_lease_renewal(self, pool_name: str, creation: _PoolCreationInFlight | None) -> None: + """Stop tracking a pool that never expires. + + Drops this call's in-flight creation slot (settling concurrent callers + waiting on it) or a renewal task started when the pool still had a lease. + """ + async with self._pools_lock: + entry = self._pools.get(pool_name) + if entry is creation or isinstance(entry, asyncio.Task): + self._pools.pop(pool_name, None) + else: + entry = None + if creation is not None: + _complete_pool_creation(creation.future) + if isinstance(entry, asyncio.Task): + entry.cancel() + with contextlib.suppress(asyncio.CancelledError): + await entry + async def _start_pool_lease_renewal(self, pool_name: str, creation: _PoolCreationInFlight) -> None: """Start lease renewal task for a pool.""" async with self._pools_lock: diff --git a/packages/sie_sdk/src/sie_sdk/client/sync.py b/packages/sie_sdk/src/sie_sdk/client/sync.py index c6ade772f..3424c8fa8 100644 --- a/packages/sie_sdk/src/sie_sdk/client/sync.py +++ b/packages/sie_sdk/src/sie_sdk/client/sync.py @@ -95,6 +95,7 @@ OutputType, PoolInfo, PoolSpec, + PoolTtl, Recommendation, RequestMetadata, ResponseInputMessage, @@ -165,6 +166,7 @@ parse_score_result, parse_terminal_json_object, parse_terminal_msgpack_object, + pool_never_expires, provisioning_retry_delay, raise_if_estimate_unroutable, raise_if_input_too_long, @@ -184,6 +186,7 @@ validate_encode_traceparent, validate_generate_grammar, validate_generate_request_body, + validate_pool_ttl, websocket_matches_base_url_origin, ) from ._sse import SSECompletion, iter_sse_payloads @@ -778,6 +781,8 @@ def _ensure_pool_created(self) -> None: request_body["minimum_worker_count"] = self._pool_spec["minimum_worker_count"] if self._pool_spec.get("pinned_models") is not None: request_body["pinned_models"] = self._pool_spec["pinned_models"] + if self._pool_spec.get("ttl_seconds") is not None: + request_body["ttl_seconds"] = self._pool_spec["ttl_seconds"] try: response = self._client.post( @@ -807,8 +812,9 @@ def _ensure_pool_created(self) -> None: self._pool_created = True - # Start lease renewal thread - self._start_lease_renewal() + # A pool that never expires has no lease to renew. + if not pool_never_expires(data): + self._start_lease_renewal() def _start_lease_renewal(self) -> None: """Start the background lease renewal thread.""" @@ -904,6 +910,7 @@ def create_pool( pinned_models: list[str] | None = None, *, queue_pool: str | None = None, + ttl_seconds: PoolTtl | None = None, ) -> None: """Create or update a resource pool for isolated capacity. @@ -922,25 +929,40 @@ def create_pool( {"l4": 4}. If omitted, all matching workers can be assigned. bundle: Optional bundle filter. When set, only workers running this bundle will be assigned to the pool. - minimum_worker_count: Per-pool warm floor (minimum machines kept warm). - The gateway emits canonical ``sie.gateway.pool.warm_floor`` - telemetry; the collector exposes ``sie_gateway_pool_warm_floor`` - to KEDA. Defaults to 0 (scale to zero). + minimum_worker_count: Per-pool warm floor: machines kept warm even with + no traffic. It is only the always-on floor: the lane autoscales + above it on demand, up to its KEDA maximum. The gateway emits + canonical ``sie.gateway.pool.warm_floor`` telemetry; the collector + exposes ``sie_gateway_pool_warm_floor`` to KEDA. Defaults to 0 + (scale to zero). pinned_models: Optional set of model ids to keep loaded so the first request to them pays no cold model-load. Each id must be a model the gateway already tracks and may be profile-qualified (``model-name:profile_name``); unknown ids are rejected. Defaults to none. queue_pool: Optional Helm/NATS queue namespace backing this logical pool. Defaults to "default", drawing from base capacity. + ttl_seconds: How long the pool lives after this client stops renewing + it. Omit for the gateway default (1200 s) on a new pool, or to keep + an existing pool's lifetime. While the client runs it renews the + pool every 60 s, so the pool lasts as long as the client plus this + grace period. Whole seconds, up to the gateway limit + (``SIE_GATEWAY_POOL_MAX_TTL_S``: by default 30 days with gateway + auth on, 1 hour without). ``"never"`` creates a pool that lives + until it is deleted, so it outlives this process and the client + does not renew it; it needs gateway auth (an admin ``api_key``). Raises: PoolError: If pool creation fails (e.g., invalid machine profile). SIEConnectionError: If unable to connect to the server. + ValueError: If ``minimum_worker_count`` is negative or ``ttl_seconds`` + is neither a positive whole number nor ``"never"``. Example: >>> client.create_pool("eval", {"l4": 2}, bundle="default", minimum_worker_count=1) >>> result = client.encode("BAAI/bge-m3", {"text": "Hello"}, gpu="eval/l4") >>> client.delete_pool("eval") + >>> # A pool for a long-lived service: no expiry, no renewal. + >>> client.create_pool("rag", {"l4": 1}, minimum_worker_count=1, ttl_seconds="never") """ with self._pools_lock: already_tracking = name in self._pools @@ -948,6 +970,7 @@ def create_pool( if minimum_worker_count is not None and minimum_worker_count < 0: msg = "minimum_worker_count must be >= 0" raise ValueError(msg) + validate_pool_ttl(ttl_seconds) logger.info( "Creating/updating pool '%s' with gpus=%s, gpu_caps=%s, bundle=%s", @@ -971,6 +994,8 @@ def create_pool( request_body["minimum_worker_count"] = minimum_worker_count if pinned_models is not None: request_body["pinned_models"] = pinned_models + if ttl_seconds is not None: + request_body["ttl_seconds"] = ttl_seconds try: response = self._client.post( @@ -997,12 +1022,24 @@ def create_pool( state = data.get("status", {}).get("state", "unknown") logger.info("Pool '%s' created/updated with state '%s'", name, state) - # Start lease renewal thread for this pool if this client is not - # already tracking it. Repeated create_pool calls intentionally still - # POST so callers can update gpus/gpu_caps on the gateway. - if not already_tracking: + # A pool that never expires has no lease: stop renewing it if this + # client was. Otherwise start renewing it if this client is not + # already. Repeated create_pool calls intentionally still POST so + # callers can update the pool on the gateway. + if pool_never_expires(data): + self._stop_pool_lease_renewal(name) + elif not already_tracking: self._start_pool_lease_renewal(name) + def _stop_pool_lease_renewal(self, pool_name: str) -> None: + """Stop this client's lease renewal thread for a pool, if any.""" + with self._pools_lock: + entry = self._pools.pop(pool_name, None) + if entry is not None: + thread, stop_event = entry + stop_event.set() + thread.join(timeout=5.0) + def _start_pool_lease_renewal(self, pool_name: str) -> None: """Start lease renewal thread for a pool.""" with self._pools_lock: @@ -1128,9 +1165,10 @@ def get_pool(self, name: str | None = None) -> PoolInfo | None: def delete_pool(self, name: str | None = None) -> bool: """Delete a pool. - This explicitly releases pool resources. Normally pools are GC'd - automatically after inactivity, so this is only needed for - immediate cleanup. + This explicitly releases pool resources. A pool with a lease is + collected automatically once it is no longer renewed, so this is only + needed for immediate cleanup. A pool created with + ``ttl_seconds="never"`` lives until it is deleted. Args: name: Pool name to delete. If None, uses the legacy constructor pool. @@ -1156,12 +1194,7 @@ def delete_pool(self, name: str | None = None) -> bool: else: return False - # Stop lease renewal thread for this pool - with self._pools_lock: - if pool_name in self._pools: - thread, stop_event = self._pools.pop(pool_name) - stop_event.set() - thread.join(timeout=5.0) + self._stop_pool_lease_renewal(pool_name) # Also handle legacy pool cleanup if this is the legacy pool if self._pool_spec is not None and pool_name == self._pool_spec.get("name"): diff --git a/packages/sie_sdk/src/sie_sdk/types.py b/packages/sie_sdk/src/sie_sdk/types.py index 1d48abe95..cc8ed35d9 100644 --- a/packages/sie_sdk/src/sie_sdk/types.py +++ b/packages/sie_sdk/src/sie_sdk/types.py @@ -1122,6 +1122,11 @@ class CapacityInfo(TypedDict, total=False): workers: list[WorkerInfo] +#: How long a pool lives after its last renewal: whole seconds, or ``"never"`` +#: for a pool that lives until it is deleted. +PoolTtl = int | Literal["never"] + + class PoolSpec(TypedDict, total=False): """Resource pool specification for creating pools. @@ -1147,6 +1152,12 @@ class PoolSpec(TypedDict, total=False): tracks (see ``GET /v1/configs/models``) and may be profile-qualified (``model-name:profile_name``); unknown ids are rejected. Defaults to an empty list (lazy-loading unchanged). + ttl_seconds: How long the pool lives after the client stops renewing it: + whole seconds (up to the gateway's ``SIE_GATEWAY_POOL_MAX_TTL_S``, by + default 30 days with gateway auth on and 1 hour without), or + ``"never"`` for a pool that lives until it is deleted and is not + renewed. ``"never"`` needs gateway auth. Defaults to the gateway's + 1200 s lease. """ name: str @@ -1156,6 +1167,7 @@ class PoolSpec(TypedDict, total=False): bundle: str minimum_worker_count: int pinned_models: list[str] + ttl_seconds: PoolTtl # --- Pool Response Types (nested structure matching gateway API) --- @@ -1204,6 +1216,9 @@ class PoolSpecResponse(TypedDict, total=False): bundle: Optional bundle filter for worker assignment. minimum_worker_count: Per-pool warm floor (minimum machines kept warm via KEDA). pinned_models: Per-pool set of model ids kept loaded (no cold model-load). + ttl_seconds: Lease in seconds, ``"never"`` for a pool that never expires (the + ``default`` pool and static Helm queue pools report it too), or ``None`` + for the gateway's default lease. """ name: str @@ -1213,6 +1228,7 @@ class PoolSpecResponse(TypedDict, total=False): bundle: str | None minimum_worker_count: int pinned_models: list[str] + ttl_seconds: PoolTtl | None class PoolResponse(TypedDict, total=False): diff --git a/packages/sie_sdk/tests/client/test_async.py b/packages/sie_sdk/tests/client/test_async.py index a7203fa79..41fd7eb90 100644 --- a/packages/sie_sdk/tests/client/test_async.py +++ b/packages/sie_sdk/tests/client/test_async.py @@ -789,6 +789,73 @@ async def test_create_pool_duplicate_posts_update_without_duplicate_renewal(self assert len(client._pools) == 1 await client.close() + @pytest.mark.asyncio + async def test_create_pool_that_never_expires_is_not_renewed(self) -> None: + resp = _make_json_response( + {"name": "rag", "spec": {"ttl_seconds": "never"}, "status": {"state": "active"}}, + status_code=201, + ) + client = SIEAsyncClient("http://localhost:8080") + client._post = AsyncMock(return_value=resp) # type: ignore + + await client.create_pool("rag", gpus={"l4": 1}, ttl_seconds="never") + + assert client._post.call_args.kwargs["json_data"]["ttl_seconds"] == "never" + assert "rag" not in client._pools + await client.close() + + @pytest.mark.asyncio + async def test_concurrent_create_of_a_pool_that_never_expires_settles_waiters(self) -> None: + release = asyncio.Event() + + async def gated_post(*_args: object, **_kwargs: object) -> _AioResponse: + await release.wait() + return _make_json_response({"name": "rag", "spec": {"ttl_seconds": "never"}, "status": {}}) + + client = SIEAsyncClient("http://localhost:8080") + client._post = gated_post # type: ignore + + first = asyncio.create_task(client.create_pool("rag", gpus={"l4": 1}, ttl_seconds="never")) + await asyncio.sleep(0) + second = asyncio.create_task(client.create_pool("rag", gpus={"l4": 1}, ttl_seconds="never")) + await asyncio.sleep(0) + release.set() + await asyncio.wait_for(asyncio.gather(first, second), timeout=5) + + assert "rag" not in client._pools + await client.close() + + @pytest.mark.asyncio + async def test_updating_a_pool_to_never_expire_cancels_its_renewal(self) -> None: + client = SIEAsyncClient("http://localhost:8080") + client._post = AsyncMock( # type: ignore + side_effect=[ + _make_json_response({"name": "rag", "spec": {"ttl_seconds": None}, "status": {}}), + _make_json_response({"name": "rag", "spec": {"ttl_seconds": "never"}, "status": {}}), + ] + ) + + await client.create_pool("rag", gpus={"l4": 1}) + renewal = client._pools["rag"] + assert isinstance(renewal, asyncio.Task) + await client.create_pool("rag", gpus={"l4": 1}, ttl_seconds="never") + + assert renewal.cancelled() + assert "rag" not in client._pools + await client.close() + + @pytest.mark.asyncio + async def test_create_pool_invalid_ttl_raises_before_any_request(self) -> None: + client = SIEAsyncClient("http://localhost:8080") + client._post = AsyncMock() # type: ignore + + with pytest.raises(ValueError, match="ttl_seconds"): + await client.create_pool("rag", gpus={"l4": 1}, ttl_seconds=0) + + client._post.assert_not_called() + assert "rag" not in client._pools + await client.close() + @pytest.mark.asyncio async def test_create_pool_connection_error_cleans_inflight_entry(self) -> None: client = SIEAsyncClient("http://localhost:8080") diff --git a/packages/sie_sdk/tests/client/test_sync.py b/packages/sie_sdk/tests/client/test_sync.py index da839634a..7a30f2430 100644 --- a/packages/sie_sdk/tests/client/test_sync.py +++ b/packages/sie_sdk/tests/client/test_sync.py @@ -80,6 +80,84 @@ def test_create_pool_duplicate_posts_update_without_duplicate_renewal(self) -> N client._pools.clear() client.close() + def test_create_pool_sends_ttl_seconds(self) -> None: + mock_response = MagicMock() + mock_response.status_code = 201 + mock_response.json.return_value = {"spec": {"ttl_seconds": 604800}, "status": {"state": "active"}} + thread = MagicMock() + + with ( + patch("sie_sdk.client.sync.httpx.Client") as mock_client, + patch("sie_sdk.client.sync.threading.Thread", return_value=thread), + ): + mock_client.return_value.post.return_value = mock_response + client = SIEClient("http://localhost:8080") + try: + client.create_pool("eval", gpus={"l4": 1}, ttl_seconds=604800) + + body = mock_client.return_value.post.call_args.kwargs["json"] + assert body == {"name": "eval", "gpus": {"l4": 1}, "ttl_seconds": 604800} + thread.start.assert_called_once() + finally: + client._pools.clear() + client.close() + + def test_create_pool_that_never_expires_is_not_renewed(self) -> None: + mock_response = MagicMock() + mock_response.status_code = 201 + mock_response.json.return_value = {"spec": {"ttl_seconds": "never"}, "status": {"state": "active"}} + + with ( + patch("sie_sdk.client.sync.httpx.Client") as mock_client, + patch("sie_sdk.client.sync.threading.Thread") as thread_cls, + ): + mock_client.return_value.post.return_value = mock_response + client = SIEClient("http://localhost:8080") + try: + client.create_pool("rag", gpus={"l4": 1}, minimum_worker_count=1, ttl_seconds="never") + + body = mock_client.return_value.post.call_args.kwargs["json"] + assert body["ttl_seconds"] == "never" + thread_cls.assert_not_called() + assert client._pools == {} + finally: + client.close() + + def test_updating_a_pool_to_never_expire_stops_its_renewal(self) -> None: + leased = MagicMock(status_code=201) + leased.json.return_value = {"spec": {"ttl_seconds": None}, "status": {"state": "active"}} + never = MagicMock(status_code=201) + never.json.return_value = {"spec": {"ttl_seconds": "never"}, "status": {"state": "active"}} + thread = MagicMock() + + with ( + patch("sie_sdk.client.sync.httpx.Client") as mock_client, + patch("sie_sdk.client.sync.threading.Thread", return_value=thread), + ): + mock_client.return_value.post.side_effect = [leased, never] + client = SIEClient("http://localhost:8080") + try: + client.create_pool("rag", gpus={"l4": 1}) + stop_event = client._pools["rag"][1] + client.create_pool("rag", gpus={"l4": 1}, ttl_seconds="never") + + assert stop_event.is_set() + thread.join.assert_called_once() + assert client._pools == {} + finally: + client.close() + + @pytest.mark.parametrize("ttl_seconds", [0, -5, 1.5, "forever", True]) + def test_create_pool_rejects_an_invalid_ttl_before_any_request(self, ttl_seconds: object) -> None: + with patch("sie_sdk.client.sync.httpx.Client") as mock_client: + client = SIEClient("http://localhost:8080") + try: + with pytest.raises(ValueError, match="ttl_seconds"): + client.create_pool("rag", gpus={"l4": 1}, ttl_seconds=ttl_seconds) # type: ignore[arg-type] + mock_client.return_value.post.assert_not_called() + finally: + client.close() + def test_trailing_slash_removed(self) -> None: """Base URL trailing slash is removed.""" with patch("sie_sdk.client.sync.httpx.Client") as mock_client: diff --git a/packages/sie_ts_sdk/src/client.ts b/packages/sie_ts_sdk/src/client.ts index 1e5acd4af..e6e8b27a0 100644 --- a/packages/sie_ts_sdk/src/client.ts +++ b/packages/sie_ts_sdk/src/client.ts @@ -148,6 +148,7 @@ import type { ModelInfo, PoolInfo, PoolSpec, + PoolTtl, RecommendOptions, Recommendation, RequestMetadata, @@ -595,6 +596,8 @@ function abortableSleep(ms: number, signal: AbortSignal): Promise { } const _LEASE_RENEWAL_MAX_RETRIES = 5; +/** `ttlSeconds` for a pool that lives until it is deleted; such a pool is never renewed. */ +const POOL_TTL_NEVER = "never"; type ItemWithWireImages = Omit & { images?: ImageWireFormat[] }; type ItemForWire = Item | ItemWithWireImages; @@ -2301,8 +2304,9 @@ export class SIEClient { * Close the client and cleanup resources. * * Stops pool lease renewal timers. Note that pools are not deleted - * automatically - they are garbage collected by the gateway after inactivity. - * This allows pool reuse if the client reconnects. + * automatically - a pool with a lease is collected by the gateway once it is + * no longer renewed, and one created with `ttlSeconds: "never"` lives until + * it is deleted. This allows pool reuse if the client reconnects. */ async close(): Promise { // Stop all pool lease renewal timers and cancel in-flight renewals @@ -2327,8 +2331,8 @@ export class SIEClient { * @param gpus - Optional machine profile requirements for pool readiness, e.g., { "l4": 2, "l4-spot": 1 } * @param gpuCaps - Optional maximum assigned workers per machine profile * @param queuePool - Optional Helm/NATS queue namespace backing this logical pool. Defaults to "default". - * @param options - Optional bundle filter, warm floor, and pinned models - * (Python SDK `create_pool` parity) + * @param options - Optional bundle filter, warm floor, pinned models, and + * TTL (Python SDK `create_pool` parity) * * @example * ```typescript @@ -2342,6 +2346,12 @@ export class SIEClient { * pinnedModels: ["BAAI/bge-m3"], * }); * + * // A pool for a long-lived service: no expiry, no renewal + * await client.createPool("rag", { l4: 1 }, undefined, undefined, { + * minimumWorkerCount: 1, + * ttlSeconds: "never", + * }); + * * // Use the pool for requests * await client.encode("BAAI/bge-m3", { text: "Hello" }, { gpu: "eval-bench/l4" }); * @@ -2361,6 +2371,16 @@ export class SIEClient { if (options.minimumWorkerCount !== undefined && options.minimumWorkerCount < 0) { throw new RangeError("minimumWorkerCount must be >= 0"); } + const { ttlSeconds } = options; + if ( + ttlSeconds !== undefined && + ttlSeconds !== POOL_TTL_NEVER && + !(Number.isInteger(ttlSeconds) && ttlSeconds >= 1) + ) { + throw new RangeError( + `ttlSeconds must be a positive whole number of seconds or "never", got ${String(ttlSeconds)}`, + ); + } // Build pool creation request const requestBody: { @@ -2371,6 +2391,7 @@ export class SIEClient { bundle?: string; minimum_worker_count?: number; pinned_models?: string[]; + ttl_seconds?: PoolTtl; } = { name, }; @@ -2392,6 +2413,9 @@ export class SIEClient { if (options.pinnedModels !== undefined) { requestBody.pinned_models = options.pinnedModels; } + if (ttlSeconds !== undefined) { + requestBody.ttl_seconds = ttlSeconds; + } const url = `${this.baseUrl}/v1/pools`; const headers: Record = { @@ -2427,6 +2451,16 @@ export class SIEClient { throw new PoolError(`Failed to create pool '${name}': ${errorMsg}`, name); } + // A pool that never expires has no lease: stop renewing it if this + // client was, and never start. + const created = (await response.json().catch(() => null)) as { + spec?: { ttl_seconds?: unknown }; + } | null; + if (created?.spec?.ttl_seconds === POOL_TTL_NEVER) { + this.stopPoolLeaseRenewal(name); + return; + } + if (alreadyTracking || this.pools.has(name)) { return; } @@ -2578,15 +2612,7 @@ export class SIEClient { * ``` */ async deletePool(name: string): Promise { - // Stop lease renewal first if we're tracking this pool - const poolState = this.pools.get(name); - if (poolState) { - if (poolState.timeoutId !== null) { - clearTimeout(poolState.timeoutId); - } - poolState.abortController.abort(); - this.pools.delete(name); - } + this.stopPoolLeaseRenewal(name); try { const url = `${this.baseUrl}/v1/pools/${encodeURIComponent(name)}`; @@ -2618,6 +2644,19 @@ export class SIEClient { } } + /** Stop this client's lease renewal for a pool, if it runs one. */ + private stopPoolLeaseRenewal(name: string): void { + const poolState = this.pools.get(name); + if (!poolState) { + return; + } + if (poolState.timeoutId !== null) { + clearTimeout(poolState.timeoutId); + } + poolState.abortController.abort(); + this.pools.delete(name); + } + private checkServerVersion(response: Response): void { if (this.versionWarningLogged) return; const serverVersion = response.headers.get(SERVER_VERSION_HEADER); diff --git a/packages/sie_ts_sdk/src/index.ts b/packages/sie_ts_sdk/src/index.ts index 309bd6d3b..35fd0e8aa 100644 --- a/packages/sie_ts_sdk/src/index.ts +++ b/packages/sie_ts_sdk/src/index.ts @@ -107,6 +107,7 @@ export type { StatusMessage, // Pool types CreatePoolOptions, + PoolTtl, PoolSpec, PoolStatus, PoolInfo, diff --git a/packages/sie_ts_sdk/src/types.ts b/packages/sie_ts_sdk/src/types.ts index c7aa45a64..e14b23483 100644 --- a/packages/sie_ts_sdk/src/types.ts +++ b/packages/sie_ts_sdk/src/types.ts @@ -655,18 +655,26 @@ export interface PoolCreateSpec { gpuCaps?: Record; } +/** + * How long a pool lives after its last renewal: whole seconds, or `"never"` + * for a pool that lives until it is deleted. + */ +export type PoolTtl = number | "never"; + /** * Optional arguments for `SIEClient.createPool`, mirroring the Python SDK's * `create_pool` keyword arguments (`bundle`, `minimum_worker_count`, - * `pinned_models`). + * `pinned_models`, `ttl_seconds`). */ export interface CreatePoolOptions { /** Optional bundle filter. When set, only workers running this bundle will be assigned to the pool. */ bundle?: string; /** - * Per-pool warm floor (minimum machines kept warm). The gateway emits - * canonical `sie.gateway.pool.warm_floor` telemetry; the collector exposes - * `sie_gateway_pool_warm_floor` to KEDA. Defaults to 0 (scale to zero). + * Per-pool warm floor: machines kept warm even with no traffic. It is only + * the always-on floor: the lane autoscales above it on demand, up to its + * KEDA maximum. The gateway emits canonical `sie.gateway.pool.warm_floor` + * telemetry; the collector exposes `sie_gateway_pool_warm_floor` to KEDA. + * Defaults to 0 (scale to zero). */ minimumWorkerCount?: number; /** @@ -675,6 +683,17 @@ export interface CreatePoolOptions { * profile-qualified ("model-name:profile_name"); unknown ids are rejected. */ pinnedModels?: string[]; + /** + * How long the pool lives after this client stops renewing it. Omit for the + * gateway default (1200 s) on a new pool, or to keep an existing pool's + * lifetime. While the client is open it renews the pool every 60 s, so the + * pool lasts as long as the client plus this grace period. Whole seconds, + * up to the gateway limit (`SIE_GATEWAY_POOL_MAX_TTL_S`: by default 30 days + * with gateway auth on, 1 hour without). `"never"` creates a pool that + * lives until it is deleted, so it outlives this process and the client + * does not renew it; it needs gateway auth (an admin `apiKey`). + */ + ttlSeconds?: PoolTtl; } /** @@ -691,8 +710,12 @@ export interface PoolSpec { gpus?: Record; /** Optional maximum assigned workers per machine profile, as returned by the gateway */ gpu_caps?: Record; - /** Optional TTL for dynamic pool leases */ - ttl_seconds?: number | null; + /** + * Lease in seconds, `"never"` for a pool that never expires (the `default` + * pool and static Helm queue pools report it too), or `null` for the + * gateway's default lease. + */ + ttl_seconds?: PoolTtl | null; /** Minimum workers kept warm for this pool */ minimum_worker_count?: number; /** Models pinned for this pool */ diff --git a/packages/sie_ts_sdk/tests/client.test.ts b/packages/sie_ts_sdk/tests/client.test.ts index e5f568ae6..41b614299 100644 --- a/packages/sie_ts_sdk/tests/client.test.ts +++ b/packages/sie_ts_sdk/tests/client.test.ts @@ -1512,6 +1512,92 @@ describe("SIEClient.createPool() - bundle / warm-floor / pinned-model args", () await client.close(); }); + + it("sends ttl_seconds on the wire", async () => { + const client = new SIEClient("http://localhost:8080"); + mockFetch.mockResolvedValueOnce(poolCreatedResponse()); + + await client.createPool("eval", { l4: 2 }, undefined, undefined, { ttlSeconds: 604_800 }); + + const [, init] = mockFetch.mock.calls[0]; + expect(JSON.parse(init.body)).toEqual({ name: "eval", gpus: { l4: 2 }, ttl_seconds: 604_800 }); + + await client.close(); + }); + + it("rejects a TTL that is neither positive whole seconds nor never", async () => { + const client = new SIEClient("http://localhost:8080"); + + for (const ttlSeconds of [0, -1, 1.5, Number.NaN]) { + await expect( + client.createPool("eval", { l4: 1 }, undefined, undefined, { ttlSeconds }), + ).rejects.toThrow(RangeError); + } + expect(mockFetch).not.toHaveBeenCalled(); + + await client.close(); + }); +}); + +describe("SIEClient.createPool() - pool lifetime and lease renewal", () => { + beforeEach(() => { + mockFetch.mockReset(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + function poolResponse(ttlSeconds: number | "never" | null): Response { + return createMockResponse({ + name: "rag", + spec: { name: "rag", ttl_seconds: ttlSeconds }, + status: { state: "active" }, + }); + } + + function renewCalls(): number { + return mockFetch.mock.calls.filter(([url]) => String(url).endsWith("/renew")).length; + } + + it("renews a pool that has a lease", async () => { + const client = new SIEClient("http://localhost:8080"); + mockFetch.mockResolvedValueOnce(poolResponse(null)); + await client.createPool("rag", { l4: 1 }); + + mockFetch.mockResolvedValue(createMockResponse({ message: "Pool renewed" })); + await vi.advanceTimersByTimeAsync(60_000); + + expect(renewCalls()).toBe(1); + await client.close(); + }); + + it("does not renew a pool that never expires", async () => { + const client = new SIEClient("http://localhost:8080"); + mockFetch.mockResolvedValueOnce(poolResponse("never")); + + await client.createPool("rag", { l4: 1 }, undefined, undefined, { ttlSeconds: "never" }); + expect(JSON.parse(mockFetch.mock.calls[0][1].body).ttl_seconds).toBe("never"); + + await vi.advanceTimersByTimeAsync(5 * 60_000); + expect(renewCalls()).toBe(0); + await client.close(); + }); + + it("stops renewing a pool once it is updated to never expire", async () => { + const client = new SIEClient("http://localhost:8080"); + mockFetch.mockResolvedValueOnce(poolResponse(null)); + await client.createPool("rag", { l4: 1 }); + + mockFetch.mockResolvedValueOnce(poolResponse("never")); + await client.createPool("rag", { l4: 1 }, undefined, undefined, { ttlSeconds: "never" }); + + mockFetch.mockResolvedValue(createMockResponse({ message: "Pool renewed" })); + await vi.advanceTimersByTimeAsync(5 * 60_000); + expect(renewCalls()).toBe(0); + await client.close(); + }); }); describe("Real-world usage patterns", () => {