diff --git a/docs/pooled-vta-claim-design.md b/docs/pooled-vta-claim-design.md new file mode 100644 index 0000000..6d9169d --- /dev/null +++ b/docs/pooled-vta-claim-design.md @@ -0,0 +1,248 @@ +# Pooled VTA Creation and Claim Design + +## 1. Product Model + +An administrator creates a batch of pooled VTAs for an event. Each VTA is a +normal Farm resource from the beginning: + +- it has its own real `users` row and database-generated user ID; +- it has its own `setup_sessions` row; +- its namespace and Vault paths are derived from that user and session; +- its DNS record, DID, PVC, configuration, Service, Ingress, and Deployment are + prepared before the event; +- its Deployment is prebuilt and waiting at zero replicas, so it uses persistent + storage but no VTA pod CPU or memory; +- its setup-session status is `pool_waiting` while it is available. + +At the station, an administrator selects an available session and displays the +ordinary VTA claim QR. Keyring scans it and shows the existing comparison code. +The administrator confirms the matching code. Confirmation writes the +claimant's admin DID through the existing mobile-connection flow, runs the +ordinary provision operation, scales the prebuilt Deployment to one replica, and +finishes at `running`. + +There is no Portal activation link, account transfer, resource-owner transfer, +or new Keyring protocol in this feature. A claimed VTA already belongs to the +user row created for it. Future Portal enrollment is a separate feature. + +## 2. Design Rule: Do Not Duplicate Existing State + +The pool schema records only batch membership. Existing tables remain the +authoritative source for everything else: + +| Concern | Source of truth | +| --- | --- | +| Account and numeric owner | `users` and `setup_sessions.user_id` | +| VTA name, image, DID, DNS record ID, domain and error | `setup_sessions` | +| Namespace | Derived by the Kubernetes client from `setup_sessions.user_id` | +| Vault path | Derived from `user_id` and `setup_session.id` | +| Cloudflare zone and load-balancer IP | Current application configuration | +| QR, scan, claimant DID and confirmation time | `mobile_connections` | +| Creation, waiting, activation, failure and running state | `setup_sessions.status` | +| Pool grouping | `pool_batches` and `pool_sessions` | + +Consequently, the pool does not store a second user ID, VTA name, namespace, +Vault target, DNS zone, ingress IP, claimant DID, claim timestamps, cleanup +cursor, or per-VTA lifecycle state. The batch keeps only a nullable timestamp +to make whole-batch cleanup durable across API restarts. + +## 3. Schema + +```sql +CREATE TABLE pool_batches ( + id bigserial PRIMARY KEY, + created_by bigint NOT NULL REFERENCES admins(id) ON DELETE RESTRICT, + label text NOT NULL, + requested_count integer NOT NULL CHECK (requested_count > 0), + idempotency_key text NOT NULL, + request_digest text NOT NULL, + cleanup_requested_at timestamptz, + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (created_by, idempotency_key) +); + +CREATE TABLE pool_sessions ( + session_id bigint PRIMARY KEY REFERENCES setup_sessions(id) ON DELETE RESTRICT, + batch_id bigint NOT NULL REFERENCES pool_batches(id) ON DELETE RESTRICT +); + +CREATE INDEX pool_sessions_batch_id ON pool_sessions(batch_id); +``` + +### `pool_batches` + +`id` identifies one batch creation request. It is not a user ID range and does +not reserve numeric user IDs. + +`created_by` records the administrator that created the batch and scopes the +idempotency key. The foreign key is restrictive so audit attribution cannot be +silently erased. + +`label` is the administrator-facing event or batch name. It has no effect on +resource identity. + +`requested_count` records the original request. The live number of members is +always counted from `pool_sessions`; it can become smaller after cleanup. +The database constraint validates only that the request is positive. The +cluster's live VTA-only estimate is the actual upper bound. + +`idempotency_key` makes a retried HTTP request return the original batch instead +of creating another 50 users and VTAs. + +`request_digest` binds the idempotency key to the original request body. Reusing +one key with a different count or configuration is rejected instead of silently +returning an unrelated batch. + +`cleanup_requested_at` is null during ordinary use. Whole-batch cleanup sets it +in the same transaction that marks every member as deleting. This is the +durable signal to delete the batch after its final member is removed. + +`created_at` is the immutable creation timestamp. There is no general mutable +batch state, so the table needs no `updated_at`. + +### `pool_sessions` + +`session_id` is both the primary key and a restrictive foreign key. A setup +session can belong to at most one pool batch. Restricting direct deletion forces +cleanup to remove external resources and pool membership deliberately before it +deletes the session and user. + +`batch_id` groups sessions for listing, selecting an available VTA, retaining +unused inventory for another event, and bulk cleanup. Its index supports those +operations. + +The table intentionally has no synthetic row ID: the setup-session ID already +uniquely identifies the pooled VTA. + +## 4. Lifecycle + +The setup session owns the lifecycle: + +| Session status | Meaning | +| --- | --- | +| `pool_creating` | User, session, and membership committed; preparation is pending | +| Existing setup stages | Namespace, Vault access, PVC, configuration and DID are being prepared | +| `pool_prepared` | DID publication completed; the VTA prebuild is being finalized | +| `pool_waiting` | The VTA is prebuilt and waiting at zero replicas, ready to claim | +| `provisioning` | A confirmed admin DID is being imported and the Deployment is starting | +| `running` | At least one Deployment replica passed readiness | +| `failed` | Preparation or activation failed; claim history determines whether cleanup is allowed | +| `deleting` | Explicit cleanup owns the session and no worker may restart it | + +The pool has no parallel state column. A list endpoint derives presentation +state by joining `pool_sessions`, `setup_sessions`, and the latest +`mobile_connections` row. + +## 5. Claim Flow + +1. Lock pool allocation and select the first session in `pool_waiting` with an + empty `admin_did`. +2. Create or return its ordinary active `mobile_connections` claim request. +3. Keyring scans the QR and submits its DID through the existing Trust Task. +4. The station displays the existing six-character comparison code. +5. Administrator confirmation sets `mobile_connections.confirmed_at`, copies + the DID into `setup_sessions.admin_did`, and advances the session to + `provisioning`. +6. The provision job imports the DID into the prepared VTA data. +7. The existing zero-replica Deployment is scaled to one. +8. Readiness advances the session to `running`, and redeem returns the VTA DID. + +The confirmed mobile connection is the durable handover record. A confirmed +activation that later fails remains claimed and cannot be allocated or cleaned +as unused inventory. + +## 6. Cleanup and Retention + +Unused inventory has no expiry. An administrator may keep it for another event, +explicitly select setup-session IDs for cleanup, or clean up an entirely unused +batch. + +Cleanup is allowed only when all of the following remain true while rows are +locked: + +- the session is a member of `pool_sessions`; +- the session is `pool_waiting` or is an unclaimed `failed` preparation; +- `setup_sessions.admin_did` is empty; +- no mobile connection is open, claimed, confirmed, provisioning, or released; +- no mobile connection has `confirmed_at` set; +- the user has no email, passkey, or full-stack access. + +Cleanup removes the resources identified by the existing session data, then +deletes the `pool_sessions` membership, setup session, and dedicated user. It +never resets PostgreSQL sequences, so deleted numeric IDs are not reused and +later users continue from the database sequence. + +Selected cleanup and `DELETE /api/v1/admin/pool-batches/{id}` use the same bulk +admission path. A session still being prepared is rejected immediately instead +of waiting for its worker; the UI disables cleanup until every selected VTA is +either prebuilt and waiting or failed. If any member fails the cleanup rules, no +member is changed. The API marks every accepted session as `deleting` and +returns `202 Accepted` without waiting for external cleanup. That session status +is the durable cleanup request for selected inventory. Whole-batch cleanup +additionally writes `cleanup_requested_at`; after background workers remove the +final member, that marker causes the empty `pool_batches` row to be deleted. + +Claimed sessions are not affected by unused-inventory cleanup. They remain +ordinary running Farm sessions managed by an administrator until a separate +Portal enrollment design is implemented. + +## 7. Failure and Retry Rules + +- Batch creation is transactional for database rows. Either all requested users, + sessions and memberships are admitted, or none are. +- External preparation is asynchronous and idempotent per setup session. +- A failure before confirmation can be retried or explicitly cleaned. +- A failure after confirmation can be retried, but never returned to inventory. +- Two station requests cannot allocate the same session: allocation serializes + pool mutations and locks the chosen setup session. +- Deployment creation at zero replicas is idempotent. Claim activation scales + that existing Deployment instead of creating a second one. + +## 8. Capacity Admission + +Batch admission uses the same strict VTA-only estimate shown by the Admin +Dashboard. A request is rejected before any batch, user, or setup-session row +is inserted when `requested_count` exceeds that estimate. For example, when +the Dashboard reports 47 additional VTA-only sessions, a request for 48 is +rejected and a request for 47 is allowed. + +Capacity is checked after an idempotency lookup, so retrying a successful +request returns its original batch even if current capacity has fallen. New +admissions first collect one cluster snapshot outside the pool transaction. +The independent Kubernetes node/pod, metrics-server, and Longhorn reads run in +parallel. After the admission lock is acquired, a database query subtracts the +current non-running pool reservations from that snapshot. This prevents two +administrators from consuming the same capacity without holding the lock during +slow cluster calls. Capacity lookup fails closed when the Dashboard cannot +produce its estimate. + +The accepted users, setup sessions, memberships, and workload-resource rows are +inserted in four bulk statements inside one transaction. Public account and VTA +identifiers are generated before their corresponding insert; a uniqueness +collision causes only the missing identifiers to be regenerated and inserted. +The committed setup-session rows supply the background runner IDs and the +`201 Created` representation, avoiding a batch lookup and detail reload on the +synchronous creation path. + +Non-running pool members are deducted from the Dashboard's VTA-only estimate. +This reserves their future CPU and memory while their Deployments are prebuilt +and waiting at zero replicas; Kubernetes pod requests alone cannot represent +that future load. The reservation remains until the session is running, when +the live Deployment requests become part of cluster statistics. + +There is intentionally no fixed product limit such as 50. The positive-value +database check protects data shape; current cluster resources decide the +operational limit. + +Batch-detail polling uses one joined query for setup sessions and their latest +claim. It interprets expired open and claimed records as lapsed without issuing +expiry updates from a GET request. The browser schedules its next poll only +after the previous request finishes, so a slow response cannot accumulate +overlapping requests. + +## 9. Migration Safety + +Migration 0043 contains the two pool tables and whole-batch cleanup marker +described above. The earlier ownership design and its `users.account_kind` and +`setup_sessions.portal_user_id` columns were rolled back and are not part of +this schema. diff --git a/internal/apidocs/openapi.yaml b/internal/apidocs/openapi.yaml index 64bf2b0..9f0e87c 100644 --- a/internal/apidocs/openapi.yaml +++ b/internal/apidocs/openapi.yaml @@ -231,6 +231,62 @@ components: [id, status, requested_count, created_count, running_count, failed_count, vta_image, created_at, updated_at, sessions] + PoolCounts: + type: object + properties: + total: { type: integer } + creating: { type: integer } + waiting: { type: integer } + claiming: { type: integer } + activating: { type: integer } + running: { type: integer } + failed: { type: integer } + deleting: { type: integer } + required: [total, creating, waiting, claiming, activating, running, failed, deleting] + + PoolSession: + type: object + properties: + session_id: { type: integer } + user_id: { type: integer } + vta_name: { type: string } + status: { type: string } + error_msg: { type: string } + url: { type: string, format: uri } + claimed: { type: boolean } + cleanable: + type: boolean + description: True only for an unclaimed session that is prebuilt and waiting or whose preparation failed + claim_state: + type: string + enum: [open, claimed, confirmed, provisioning, released, rejected, lapsed] + required: [session_id, user_id, vta_name, status, url, claimed, cleanable] + + PoolBatch: + type: object + properties: + id: { type: integer } + label: { type: string } + requested_count: { type: integer } + vta_image: { type: string } + created_at: { type: string, format: date-time } + counts: { $ref: "#/components/schemas/PoolCounts" } + sessions: + type: array + items: { $ref: "#/components/schemas/PoolSession" } + required: [id, label, requested_count, created_at, counts] + + PoolClaim: + type: object + properties: + batch_id: { type: integer } + session_id: { type: integer } + vta_name: { type: string } + enabled: { type: boolean } + server_time: { type: string, format: date-time } + connection: { $ref: "#/components/schemas/MobileConnection" } + required: [batch_id, session_id, vta_name, enabled, server_time, connection] + RateLimited: type: object properties: @@ -1627,6 +1683,320 @@ paths: "503": description: Kubernetes is not configured + /api/v1/admin/pool-batches: + get: + summary: List pooled VTA batches + description: Returns batch summaries with counts derived from their setup sessions and latest claim records. + tags: [Admin] + security: + - CookieAuthAdmin: [] + responses: + "200": + description: Pool batch summaries + content: + application/json: + schema: + type: array + items: { $ref: "#/components/schemas/PoolBatch" } + post: + summary: Create a pooled VTA batch + description: | + Atomically creates one dedicated user and setup session per requested + VTA with bulk database writes, then queues durable background + preparation. The slow cluster snapshot is collected before serialized + database admission; after acquiring the admission lock, the API + rechecks current pool reservations and rejects the request before any + rows are inserted when `count` exceeds the remaining strict VTA-only + capacity shown by the Admin Dashboard. The 201 response is assembled + directly from the committed batch instead of waiting for a second + detail query. + + Repeating an accepted request with the same `Idempotency-Key` and body + returns the original batch. Reusing the key with a different body is a + conflict. + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: header + name: Idempotency-Key + required: true + schema: { type: string, minLength: 1, maxLength: 128 } + requestBody: + required: true + content: + application/json: + schema: + type: object + properties: + label: { type: string, minLength: 1, maxLength: 120 } + count: { type: integer, minimum: 1 } + vta_image: { type: string, minLength: 1 } + required: [label, count, vta_image] + responses: + "201": + description: Batch admitted and queued + content: + application/json: + schema: { $ref: "#/components/schemas/PoolBatch" } + "200": + description: Idempotent replay of an existing batch + content: + application/json: + schema: { $ref: "#/components/schemas/PoolBatch" } + "409": + description: Idempotency key conflicts with another request + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + "422": + description: Requested count exceeds live VTA-only capacity + content: + application/json: + schema: + allOf: + - { $ref: "#/components/schemas/Error" } + - type: object + properties: + reason: { type: string, const: insufficient_capacity } + available: { type: integer, minimum: 0 } + + /api/v1/admin/pool-batches/{id}: + get: + summary: Get one pooled VTA batch + description: Reads every setup session and its latest claim in one joined query. Expired open or claimed records are represented as lapsed without mutating claim rows during this GET. + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: path + name: id + required: true + schema: { type: integer, minimum: 1 } + responses: + "200": + description: Batch details and every remaining member + content: + application/json: + schema: { $ref: "#/components/schemas/PoolBatch" } + "404": + description: Batch not found + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + delete: + summary: Clean up an entirely unused pooled VTA batch + description: | + Uses bulk database checks to verify that every remaining member is + unused and has finished preparation. It atomically marks every member + for durable cleanup and returns without waiting for external resource + deletion. Background workers remove the resources and delete the batch + after its final member. The whole request is rejected immediately if + any member is still preparing, claimed, has an active claim, or has + acquired login credentials. + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: path + name: id + required: true + schema: { type: integer, minimum: 1 } + responses: + "202": + description: Durable whole-batch cleanup accepted; background workers resume it after restarts + content: + application/json: + schema: + type: object + properties: + batch_id: { type: integer, minimum: 1 } + session_ids: + type: array + items: { type: integer, minimum: 1 } + status: { type: string, const: deleting } + required: [batch_id, session_ids, status] + "204": + description: The empty batch was deleted immediately + "404": + description: Batch not found + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + "409": + description: At least one member is still preparing, claimed, active, or no longer cleanable + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + + /api/v1/admin/pool-batches/{id}/claims: + post: + summary: Reserve the next waiting VTA and create its Keyring QR + description: Concurrent calls never receive the same active setup session. The returned trigger link is rendered as the QR payload. + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: path + name: id + required: true + schema: { type: integer, minimum: 1 } + responses: + "200": + description: Claim QR reservation + content: + application/json: + schema: { $ref: "#/components/schemas/PoolClaim" } + "409": + description: No waiting VTA is available + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + + /api/v1/admin/pool-claims/{request_id}: + get: + summary: Poll a pooled VTA claim + description: Reconciles QR and confirmation expiry before returning the latest state. + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: path + name: request_id + required: true + schema: { type: string, format: uuid } + responses: + "200": + description: Current claim state + content: + application/json: + schema: { $ref: "#/components/schemas/PoolClaim" } + "404": + description: Claim not found + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + + /api/v1/admin/pool-claims/{request_id}/confirm: + post: + summary: Confirm matching Keyring comparison codes and activate the VTA + description: Confirmation permanently protects this VTA from unused-inventory cleanup and starts the prebuilt VTA that was waiting at zero replicas. + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: path + name: request_id + required: true + schema: { type: string, format: uuid } + responses: + "200": + description: Confirmed or provisioning claim + content: + application/json: + schema: { $ref: "#/components/schemas/PoolClaim" } + "409": + description: Claim is not confirmable + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + "410": + description: Confirmation window expired + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + + /api/v1/admin/pool-claims/{request_id}/reject: + post: + summary: Cancel an open QR or reject a mismatched pooled VTA claim + description: An open request becomes lapsed; a claimed request becomes rejected. The untouched VTA returns to the batch inventory. + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: path + name: request_id + required: true + schema: { type: string, format: uuid } + responses: + "200": + description: Cancelled or rejected claim; its untouched VTA may be offered again + content: + application/json: + schema: { $ref: "#/components/schemas/PoolClaim" } + "409": + description: Claim is no longer open or rejectable + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + + /api/v1/admin/pool-sessions/cleanup: + post: + summary: Queue cleanup for selected unused pooled VTAs + description: The whole selection is checked and marked deleting with bulk database operations. The request is rejected immediately if any selected session is still preparing, claimed, actively being claimed, or has acquired login credentials. A 202 response means the durable deleting state has committed; background workers perform and retry external resource deletion. + tags: [Admin] + security: + - CookieAuthAdmin: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + properties: + session_ids: + type: array + minItems: 1 + uniqueItems: true + items: { type: integer, minimum: 1 } + required: [session_ids] + responses: + "202": + description: Durable cleanup queued; external deletion continues in background workers and retries after restarts + content: + application/json: + schema: + type: object + properties: + session_ids: + type: array + items: { type: integer } + status: { type: string, const: deleting } + required: [session_ids, status] + "409": + description: At least one selected session is still preparing or otherwise protected + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + + /api/v1/admin/pool-sessions/{id}/retry: + post: + summary: Retry preparation of an unclaimed failed pooled VTA + tags: [Admin] + security: + - CookieAuthAdmin: [] + parameters: + - in: path + name: id + required: true + schema: { type: integer, minimum: 1 } + responses: + "202": + description: Preparation requeued + content: + application/json: + schema: + type: object + properties: + session_id: { type: integer } + status: { type: string, const: pool_creating } + required: [session_id, status] + "409": + description: Session is claimed or is not a retryable failed pool member + content: + application/json: + schema: { $ref: "#/components/schemas/Error" } + /api/v1/admin/load-tests: post: summary: Start a VTA provisioning load test @@ -1943,7 +2313,7 @@ paths: "400": { description: Invalid JSON body } "401": { description: Admin authentication required } "403": { description: Insufficient role } - "422": { description: Missing, duplicate, or unknown component, or invalid memory values } + "422": { description: "Missing, duplicate, or unknown component, or invalid memory values" } "500": { description: Could not save defaults } /api/v1/admin/setup-sessions/resources: @@ -2009,7 +2379,7 @@ paths: error: { type: string } "404": { description: One or more sessions were not found } "409": { description: One or more sessions are not running } - "422": { description: Validation, capacity, or Kubernetes dry-run failed } + "422": { description: "Validation, capacity, or Kubernetes dry-run failed" } "503": { description: Kubernetes is not configured } /api/v1/admin/setup-sessions/{id}/resources: @@ -5759,7 +6129,7 @@ paths: '400': {description: Malformed document or payload} '409': {description: Document ID reused with different content} '415': {description: Content-Type is not application/json} - '422': {description: Task, proof, recipient, expiry, or business-policy rejection} + '422': {description: "Task, proof, recipient, expiry, or business-policy rejection"} '429': {description: Rate limited} '503': {description: Trust Task service or replay record unavailable} diff --git a/internal/cloudflare/client.go b/internal/cloudflare/client.go index bad6f47..71f3144 100644 --- a/internal/cloudflare/client.go +++ b/internal/cloudflare/client.go @@ -8,6 +8,7 @@ import ( "fmt" "net" "net/http" + "net/url" ) const baseURL = "https://api.cloudflare.com/client/v4" @@ -121,6 +122,47 @@ func (c *Client) CreateARecord(ctx context.Context, name, ip string) (string, er return result.Result.ID, nil } +// EnsureARecord returns an existing matching record or creates it. Pool builds +// use this instead of a blind create so a process crash after Cloudflare +// accepts the record but before its ID reaches PostgreSQL is recoverable. +func (c *Client) EnsureARecord(ctx context.Context, name, ip string) (string, error) { + endpoint, err := url.Parse(fmt.Sprintf("%s/zones/%s/dns_records", baseURL, c.zoneID)) + if err != nil { + return "", err + } + query := endpoint.Query() + query.Set("type", "A") + query.Set("name", name) + endpoint.RawQuery = query.Encode() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint.String(), nil) + if err != nil { + return "", err + } + resp, err := c.do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + var result apiResponse[[]DNSRecord] + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + if err := cfError(result.Success, result.Errors); err != nil { + return "", err + } + if len(result.Result) > 1 { + return "", fmt.Errorf("cloudflare: multiple A records exist for %s", name) + } + if len(result.Result) == 1 { + record := result.Result[0] + if record.Content != ip || !record.Proxied { + return "", fmt.Errorf("cloudflare: A record for %s has unexpected target or proxy mode", name) + } + return record.ID, nil + } + return c.CreateARecord(ctx, name, ip) +} + // DeleteRecord removes the DNS record with the given Cloudflare record ID. // Returns nil if the record does not exist (idempotent). func (c *Client) DeleteRecord(ctx context.Context, recordID string) error { diff --git a/internal/connection/store.go b/internal/connection/store.go index c97ce52..d6e74c8 100644 --- a/internal/connection/store.go +++ b/internal/connection/store.go @@ -91,7 +91,7 @@ func Ready(s *model.SetupSession) bool { if s.IsFullStack() { return s.Status == "awaiting_admin_did" } - return s.Status == "vta_setup_complete" + return s.Status == "vta_setup_complete" || s.Status == model.StatusPoolWaiting } func lockSession(tx *gorm.DB, id uint) (*model.SetupSession, error) { @@ -143,6 +143,16 @@ func finalStatus(status string) bool { // Change creates, refreshes, or cancels an owner-visible reservation. It always // locks the setup session first, matching claim/confirm/provision transitions. func Change(ctx context.Context, db *gorm.DB, sessionID uint, action, expectedID string) (*model.MobileConnection, error) { + return change(ctx, db, sessionID, action, expectedID, false) +} + +// ChangePool is called inside the pool allocation transaction, after the slot +// is reserved. Public owner routes cannot allocate prebuilt pool inventory. +func ChangePool(ctx context.Context, db *gorm.DB, sessionID uint) (*model.MobileConnection, error) { + return change(ctx, db, sessionID, "create", "", true) +} + +func change(ctx context.Context, db *gorm.DB, sessionID uint, action, expectedID string, pooled bool) (*model.MobileConnection, error) { var result *model.MobileConnection err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { s, err := lockSession(tx, sessionID) @@ -177,6 +187,10 @@ func Change(ctx context.Context, db *gorm.DB, sessionID uint, action, expectedID if err != nil { return err } + isPool := s.Status == model.StatusPoolWaiting + if isPool != pooled || isPool && operation != OperationProvisionVTA { + return ErrConflict + } if current != nil { switch current.Status { case "claimed", "confirmed", "provisioning": @@ -403,6 +417,9 @@ func StartProvisioning(ctx context.Context, db *gorm.DB, requestID string) (*mod } switch result.Operation { case OperationProvisionVTA: + if s.Status == model.StatusPoolWaiting && result.ConfirmedAt == nil { + return ErrConflict + } if err := beginProvision(tx, s, result.AdminDid, now); err != nil { return err } @@ -491,6 +508,9 @@ func AcceptManual(ctx context.Context, db *gorm.DB, sessionID uint, value string if err != nil { return err } + if s.Status == model.StatusPoolWaiting { + return ErrConflict + } current, err := Current(tx, sessionID) if err != nil { return err diff --git a/internal/handler/admin_enroll.go b/internal/handler/admin_enroll.go index 4a6bb68..ae17aa1 100644 --- a/internal/handler/admin_enroll.go +++ b/internal/handler/admin_enroll.go @@ -10,6 +10,7 @@ import ( "github.com/ic3software/vtafarm-api/internal/middleware" "github.com/ic3software/vtafarm-api/internal/model" + "github.com/ic3software/vtafarm-api/internal/publicid" ) type AdminEnrollHandler struct { @@ -70,7 +71,10 @@ func (h *AdminEnrollHandler) Enroll(c *gin.Context) { const maxAttempts = 5 var createErr error for range maxAttempts { - admin.UniqueId = generateUniqueId() + admin.UniqueId, createErr = publicid.NewAccount() + if createErr != nil { + break + } createErr = h.db.Create(&admin).Error if createErr == nil { break diff --git a/internal/handler/admin_pool.go b/internal/handler/admin_pool.go new file mode 100644 index 0000000..0a6a21a --- /dev/null +++ b/internal/handler/admin_pool.go @@ -0,0 +1,608 @@ +package handler + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" + + "github.com/ic3software/vtafarm-api/internal/connection" + "github.com/ic3software/vtafarm-api/internal/k8s" + "github.com/ic3software/vtafarm-api/internal/middleware" + "github.com/ic3software/vtafarm-api/internal/model" + poolstore "github.com/ic3software/vtafarm-api/internal/pool" + "github.com/ic3software/vtafarm-api/internal/resourceprofile" + "github.com/ic3software/vtafarm-api/internal/setup" +) + +type PoolHandler struct { + db *gorm.DB + setup *SetupHandler + dashboard *DashboardHandler + mobile *MobileConnectionHandler + runner *poolstore.Runner +} + +func NewPoolHandler( + db *gorm.DB, + setupHandler *SetupHandler, + dashboard *DashboardHandler, + mobile *MobileConnectionHandler, +) *PoolHandler { + h := &PoolHandler{db: db, setup: setupHandler, dashboard: dashboard, mobile: mobile} + var build func(context.Context, uint) error + if setupHandler != nil && setupHandler.orch != nil { + build = setupHandler.orch.BuildPool + } + h.runner = poolstore.NewRunner(db, h.prepareSession, build, h.cleanupSession) + return h +} + +func (h *PoolHandler) Start(ctx context.Context) { + h.runner.Start(ctx) +} + +type createPoolBatchRequest struct { + Label string `json:"label"` + Count int `json:"count"` + VtaImage string `json:"vta_image"` +} + +type poolCounts struct { + Total int `json:"total"` + Creating int `json:"creating"` + Waiting int `json:"waiting"` + Claiming int `json:"claiming"` + Activating int `json:"activating"` + Running int `json:"running"` + Failed int `json:"failed"` + Deleting int `json:"deleting"` +} + +type poolSessionItem struct { + SessionID uint `json:"session_id"` + UserID uint `json:"user_id"` + VtaName string `json:"vta_name"` + Status string `json:"status"` + ErrorMsg string `json:"error_msg,omitempty"` + URL string `json:"url"` + Claimed bool `json:"claimed"` + Cleanable bool `json:"cleanable"` + ClaimState string `json:"claim_state,omitempty"` +} + +type poolBatchView struct { + ID uint `json:"id"` + Label string `json:"label"` + RequestedCount int `json:"requested_count"` + VtaImage string `json:"vta_image,omitempty"` + CreatedAt time.Time `json:"created_at"` + Counts poolCounts `json:"counts"` + Sessions []poolSessionItem `json:"sessions,omitempty"` +} + +type poolClaimView struct { + BatchID uint `json:"batch_id"` + SessionID uint `json:"session_id"` + VtaName string `json:"vta_name"` + Enabled bool `json:"enabled"` + ServerTime time.Time `json:"server_time"` + Connection *mobileView `json:"connection"` +} + +// CreateBatch — POST /api/v1/admin/pool-batches. +func (h *PoolHandler) CreateBatch(c *gin.Context) { + if h.setup == nil || h.setup.cf == nil || h.setup.k8s == nil || h.setup.orch == nil || + h.setup.didHosting == nil || h.dashboard == nil || h.mobile == nil || !h.mobile.enabled { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "pooled VTA creation is not fully configured"}) + return + } + if h.setup.ingressIP == "" || h.setup.clusterDomain == "" { + c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "cluster not configured: CLUSTER_INGRESS_IP and CLUSTER_DOMAIN must be set"}) + return + } + var request createPoolBatchRequest + if err := c.ShouldBindJSON(&request); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + request.Label = strings.TrimSpace(request.Label) + request.VtaImage = strings.TrimSpace(request.VtaImage) + key := strings.TrimSpace(c.GetHeader("Idempotency-Key")) + if request.Label == "" || len(request.Label) > 120 { + c.JSON(http.StatusBadRequest, gin.H{"error": "label must be between 1 and 120 characters"}) + return + } + if request.Count < 1 { + c.JSON(http.StatusBadRequest, gin.H{"error": "count must be at least 1"}) + return + } + if request.VtaImage == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "vta_image is required"}) + return + } + if key == "" || len(key) > 128 { + c.JSON(http.StatusBadRequest, gin.H{"error": "Idempotency-Key must be between 1 and 128 characters"}) + return + } + + digestBody, _ := json.Marshal(request) + digestBytes := sha256.Sum256(digestBody) + digest := hex.EncodeToString(digestBytes[:]) + adminID := c.MustGet(middleware.ContextUserID).(uint) + if existing, lookupErr := poolstore.LookupBatch(c.Request.Context(), h.db, adminID, key, digest); lookupErr == nil { + view, viewErr := h.batchView(c.Request.Context(), existing.ID, true) + if viewErr != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "batch exists but could not be loaded"}) + return + } + c.JSON(http.StatusOK, view) + return + } else if !errors.Is(lookupErr, gorm.ErrRecordNotFound) { + if errors.Is(lookupErr, poolstore.ErrConflict) { + c.JSON(http.StatusConflict, gin.H{"error": "Idempotency-Key was already used with a different request"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to check pool batch request"}) + return + } + + var infra sharedInfra + var reason, detail string + var profiles resourceprofile.Profiles + var profileErr error + var clusterStats *k8s.ClusterStats + var clusterErr error + var reads sync.WaitGroup + reads.Add(3) + go func() { + defer reads.Done() + infra, _, reason, detail = h.setup.resolvePlatformInfra() + }() + go func() { + defer reads.Done() + profiles, profileErr = resourceprofile.Load(c.Request.Context(), h.db) + }() + go func() { + defer reads.Done() + clusterStats, clusterErr = h.dashboard.clusterCapacitySnapshot(c.Request.Context()) + }() + reads.Wait() + if reason != "" { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": detail, "reason": reason}) + return + } + if profileErr != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "could not load resource defaults"}) + return + } + if clusterErr != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "could not read cluster capacity"}) + return + } + profile := profiles[k8s.ComponentVTA] + capacitySnapshot := h.dashboard.vtaOnlyCapacityFromSnapshot(clusterStats, profiles) + admitted, err := poolstore.Admit(c.Request.Context(), h.db, poolstore.Admission{ + AdminID: adminID, + Key: key, + Digest: digest, + Label: request.Label, + Count: request.Count, + Capacity: capacitySnapshot, + Session: model.SetupSession{ + Mode: model.ModeVtaOnly, + DomainType: model.DomainManaged, + Domain: h.setup.clusterDomain, + Subdomain: setup.EnvPrefix(h.setup.appEnv) + "vta-", + MediatorDid: infra.MediatorDid, + DidHostingServerURL: strings.TrimSuffix(infra.ServerURL, "/"), + DidHostingControlURL: infra.ControlURL, + DIDHostingDid: infra.DaemonDid, + ConnectionSource: model.ConnectionPlatform, + VtaImage: request.VtaImage, + Portable: true, + PreRotationCount: 1, + }, + MemoryRequest: profile.MemoryRequest, + MemoryLimit: profile.MemoryLimit, + }) + if err != nil { + var capacityErr *poolstore.CapacityError + switch { + case errors.As(err, &capacityErr): + c.JSON(http.StatusUnprocessableEntity, gin.H{ + "error": fmt.Sprintf("the cluster currently has capacity for %d additional VTA-only sessions", capacityErr.Available), + "reason": "insufficient_capacity", "available": capacityErr.Available, + }) + case errors.Is(err, poolstore.ErrConflict): + c.JSON(http.StatusConflict, gin.H{"error": "Idempotency-Key was already used with a different request"}) + default: + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create pool batch"}) + } + return + } + for i := range admitted.Sessions { + h.runner.Kick(admitted.Sessions[i].ID) + } + status := http.StatusOK + var view *poolBatchView + if admitted.Created { + status = http.StatusCreated + view = buildPoolBatchView(*admitted.Batch, admitted.Sessions, nil, true) + } else { + view, err = h.batchView(c.Request.Context(), admitted.Batch.ID, true) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "batch exists but could not be loaded"}) + return + } + } + c.JSON(status, view) +} + +func (h *PoolHandler) ListBatches(c *gin.Context) { + var batches []model.PoolBatch + if err := h.db.WithContext(c.Request.Context()).Order("id DESC").Find(&batches).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list pool batches"}) + return + } + views := make([]poolBatchView, 0, len(batches)) + for _, batch := range batches { + view, err := h.batchView(c.Request.Context(), batch.ID, false) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load pool batches"}) + return + } + views = append(views, *view) + } + c.JSON(http.StatusOK, views) +} + +func (h *PoolHandler) GetBatch(c *gin.Context) { + batchID, ok := positiveID(c.Param("id")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid pool batch ID"}) + return + } + view, err := h.batchView(c.Request.Context(), batchID, true) + if errors.Is(err, gorm.ErrRecordNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "pool batch not found"}) + return + } + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load pool batch"}) + return + } + c.JSON(http.StatusOK, view) +} + +func (h *PoolHandler) Reserve(c *gin.Context) { + if h.mobile == nil || !h.mobile.enabled { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "VTA claim is not available", "reason": "vta_claim_disabled"}) + return + } + batchID, ok := positiveID(c.Param("id")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid pool batch ID"}) + return + } + var exists int64 + if err := h.db.Model(&model.PoolBatch{}).Where("id = ?", batchID).Count(&exists).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load pool batch"}) + return + } + if exists == 0 { + c.JSON(http.StatusNotFound, gin.H{"error": "pool batch not found"}) + return + } + claim, err := poolstore.Reserve(c.Request.Context(), h.db, batchID) + if err != nil { + if errors.Is(err, poolstore.ErrEmpty) { + c.JSON(http.StatusConflict, gin.H{"error": "no waiting VTA is available in this batch", "reason": "pool_empty"}) + return + } + writeConnectionError(c, err) + return + } + h.writeClaim(c, claim.ID) +} + +func (h *PoolHandler) GetClaim(c *gin.Context) { + h.writeClaim(c, c.Param("request_id")) +} + +func (h *PoolHandler) ConfirmClaim(c *gin.Context) { + claim, err := poolstore.Confirm(c.Request.Context(), h.db, c.Param("request_id")) + if err != nil { + writeConnectionError(c, err) + return + } + if h.setup != nil && h.setup.orch != nil { + h.setup.orch.KickProvision(claim.SessionID) + } + h.writeClaim(c, claim.ID) +} + +func (h *PoolHandler) RejectClaim(c *gin.Context) { + requestID := c.Param("request_id") + if err := poolstore.Reject(c.Request.Context(), h.db, requestID); err != nil { + writeConnectionError(c, err) + return + } + h.writeClaim(c, requestID) +} + +func (h *PoolHandler) Cleanup(c *gin.Context) { + var request struct { + SessionIDs []uint `json:"session_ids"` + } + if err := c.ShouldBindJSON(&request); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if err := poolstore.QueueCleanup(c.Request.Context(), h.db, request.SessionIDs); err != nil { + if errors.Is(err, poolstore.ErrConflict) || errors.Is(err, connection.ErrConflict) { + c.JSON(http.StatusConflict, gin.H{"error": "one or more sessions are still preparing, claimed, active, or no longer cleanable"}) + return + } + writeConnectionError(c, err) + return + } + for _, sessionID := range request.SessionIDs { + h.runner.Kick(sessionID) + } + c.JSON(http.StatusAccepted, gin.H{"session_ids": request.SessionIDs, "status": "deleting"}) +} + +func (h *PoolHandler) CleanupBatch(c *gin.Context) { + batchID, ok := positiveID(c.Param("id")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid pool batch ID"}) + return + } + sessionIDs, err := poolstore.QueueBatchCleanup(c.Request.Context(), h.db, batchID) + if errors.Is(err, gorm.ErrRecordNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "pool batch not found"}) + return + } + if errors.Is(err, poolstore.ErrConflict) || errors.Is(err, connection.ErrConflict) { + c.JSON(http.StatusConflict, gin.H{"error": "one or more sessions are still preparing, claimed, active, or no longer cleanable"}) + return + } + if err != nil { + writeConnectionError(c, err) + return + } + if len(sessionIDs) == 0 { + c.Status(http.StatusNoContent) + return + } + for _, sessionID := range sessionIDs { + h.runner.Kick(sessionID) + } + c.JSON(http.StatusAccepted, gin.H{"batch_id": batchID, "session_ids": sessionIDs, "status": "deleting"}) +} + +func (h *PoolHandler) Retry(c *gin.Context) { + sessionID, ok := positiveID(c.Param("id")) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid pool session ID"}) + return + } + if err := poolstore.RetryBuild(c.Request.Context(), h.db, sessionID); err != nil { + if errors.Is(err, poolstore.ErrConflict) { + c.JSON(http.StatusConflict, gin.H{"error": "only an unclaimed failed pool session can be retried"}) + return + } + writeConnectionError(c, err) + return + } + h.runner.Kick(sessionID) + c.JSON(http.StatusAccepted, gin.H{"session_id": sessionID, "status": model.StatusPoolCreating}) +} + +func (h *PoolHandler) batchView(ctx context.Context, batchID uint, includeSessions bool) (*poolBatchView, error) { + var batch model.PoolBatch + if err := h.db.WithContext(ctx).First(&batch, batchID).Error; err != nil { + return nil, err + } + type sessionRow struct { + model.SetupSession `gorm:"embedded"` + ClaimStatus *string `gorm:"column:claim_status"` + ClaimConfirmedAt *time.Time `gorm:"column:claim_confirmed_at"` + ClaimExpiresAt *time.Time `gorm:"column:claim_expires_at"` + ClaimConfirmBy *time.Time `gorm:"column:claim_confirm_by"` + ReadAt time.Time `gorm:"column:read_at"` + } + var rows []sessionRow + if err := h.db.WithContext(ctx).Raw(` + SELECT s.*, + claim.status AS claim_status, + claim.confirmed_at AS claim_confirmed_at, + claim.expires_at AS claim_expires_at, + claim.confirm_by AS claim_confirm_by, + clock_timestamp() AS read_at + FROM setup_sessions s + JOIN pool_sessions ps ON ps.session_id = s.id + LEFT JOIN LATERAL ( + SELECT mc.status, mc.confirmed_at, mc.expires_at, mc.confirm_by + FROM mobile_connections mc + WHERE mc.session_id = s.id + ORDER BY mc.created_at DESC, mc.id DESC + LIMIT 1 + ) claim ON true + WHERE ps.batch_id = ? + ORDER BY s.id`, batchID).Scan(&rows).Error; err != nil { + return nil, err + } + sessions := make([]model.SetupSession, len(rows)) + latest := make(map[uint]poolClaimSummary, len(rows)) + for i := range rows { + sessions[i] = rows[i].SetupSession + if rows[i].ClaimStatus == nil { + continue + } + status := *rows[i].ClaimStatus + if status == "open" && rows[i].ClaimExpiresAt != nil && !rows[i].ClaimExpiresAt.After(rows[i].ReadAt) { + status = "lapsed" + } + if status == "claimed" && rows[i].ClaimConfirmBy != nil && !rows[i].ClaimConfirmBy.After(rows[i].ReadAt) { + status = "lapsed" + } + latest[rows[i].ID] = poolClaimSummary{Status: status, ConfirmedAt: rows[i].ClaimConfirmedAt} + } + return buildPoolBatchView(batch, sessions, latest, includeSessions), nil +} + +type poolClaimSummary struct { + Status string + ConfirmedAt *time.Time +} + +func buildPoolBatchView(batch model.PoolBatch, sessions []model.SetupSession, latest map[uint]poolClaimSummary, includeSessions bool) *poolBatchView { + view := &poolBatchView{ + ID: batch.ID, Label: batch.Label, RequestedCount: batch.RequestedCount, + CreatedAt: batch.CreatedAt, Sessions: []poolSessionItem{}, + } + if len(sessions) > 0 { + view.VtaImage = sessions[0].VtaImage + } + for i := range sessions { + session := &sessions[i] + current, hasClaim := latest[session.ID] + item := poolSessionItem{ + SessionID: session.ID, UserID: session.UserID, VtaName: session.VtaName, + Status: session.Status, ErrorMsg: session.ErrorMsg, URL: session.PublicURL(), + } + if hasClaim { + item.ClaimState = current.Status + } + item.Claimed = session.AdminDid != "" || hasClaim && current.ConfirmedAt != nil + activeClaim := hasClaim && (current.Status == "open" || current.Status == "claimed" || current.Status == "confirmed" || current.Status == "provisioning") + cleanupReady := session.Status == model.StatusPoolWaiting || session.Status == "failed" + item.Cleanable = cleanupReady && !item.Claimed && !activeClaim + view.Counts.Total++ + switch { + case session.Status == "deleting": + view.Counts.Deleting++ + case session.Status == "running": + view.Counts.Running++ + case session.Status == "failed": + view.Counts.Failed++ + case item.Claimed || session.Status == "provisioning": + view.Counts.Activating++ + case activeClaim: + view.Counts.Claiming++ + case session.Status == model.StatusPoolWaiting: + view.Counts.Waiting++ + default: + view.Counts.Creating++ + } + if includeSessions { + view.Sessions = append(view.Sessions, item) + } + } + return view +} + +func (h *PoolHandler) writeClaim(c *gin.Context, requestID string) { + var claim model.MobileConnection + if err := h.db.WithContext(c.Request.Context()).First(&claim, "id = ?", requestID).Error; err != nil { + writeConnectionError(c, err) + return + } + var membership model.PoolSession + if err := h.db.WithContext(c.Request.Context()).First(&membership, "session_id = ?", claim.SessionID).Error; err != nil { + writeConnectionError(c, err) + return + } + var session model.SetupSession + if err := h.db.WithContext(c.Request.Context()).First(&session, claim.SessionID).Error; err != nil { + writeConnectionError(c, err) + return + } + if err := connection.Reconcile(c.Request.Context(), h.db, &claim, &session); err != nil { + writeConnectionError(c, err) + return + } + now, err := connection.Now(h.db.WithContext(c.Request.Context())) + if err != nil { + writeConnectionError(c, err) + return + } + c.JSON(http.StatusOK, poolClaimView{ + BatchID: membership.BatchID, SessionID: session.ID, VtaName: session.VtaName, + Enabled: h.mobile != nil && h.mobile.enabled, ServerTime: now, + Connection: h.mobile.view(&claim), + }) +} + +func (h *PoolHandler) prepareSession(ctx context.Context, sessionID uint) error { + if h.setup == nil || h.setup.cf == nil { + return errors.New("cloudflare is not configured") + } + var session model.SetupSession + if err := h.db.WithContext(ctx).First(&session, sessionID).Error; err != nil { + return err + } + if !poolstore.IsSession(h.db.WithContext(ctx), sessionID) || session.Status == "deleting" { + return poolstore.ErrConflict + } + if session.CFRecordID != "" { + return nil + } + recordID, err := h.setup.cf.EnsureARecord(ctx, session.FQDN(), h.setup.ingressIP) + if err != nil { + return fmt.Errorf("failed to create DNS record: %w", err) + } + updated := h.db.WithContext(ctx).Model(&model.SetupSession{}). + Where("id = ? AND cf_record_id = '' AND status <> 'deleting'", sessionID). + Update("cf_record_id", recordID) + if updated.Error != nil { + return updated.Error + } + if updated.RowsAffected == 0 { + var current model.SetupSession + if err := h.db.WithContext(ctx).Select("cf_record_id", "status").First(¤t, sessionID).Error; err != nil { + return err + } + if current.Status == "deleting" { + return poolstore.ErrConflict + } + if current.CFRecordID == "" { + return errors.New("DNS record ownership changed during pool preparation") + } + } + return nil +} + +func (h *PoolHandler) cleanupSession(ctx context.Context, sessionID uint) error { + if h.setup == nil { + return errors.New("setup cleanup is not configured") + } + var membership model.PoolSession + if err := h.db.WithContext(ctx).First(&membership, "session_id = ?", sessionID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil + } + return err + } + if err := h.setup.cleanupPoolSession(ctx, sessionID); err != nil { + return err + } + return poolstore.FinalizeBatchCleanup(ctx, h.db, membership.BatchID) +} + +func positiveID(value string) (uint, bool) { + id, err := strconv.ParseUint(value, 10, 64) + return uint(id), err == nil && id > 0 +} diff --git a/internal/handler/admin_pool_test.go b/internal/handler/admin_pool_test.go new file mode 100644 index 0000000..0bb5562 --- /dev/null +++ b/internal/handler/admin_pool_test.go @@ -0,0 +1,128 @@ +package handler + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + "github.com/ic3software/vtafarm-api/internal/model" + poolstore "github.com/ic3software/vtafarm-api/internal/pool" + "github.com/ic3software/vtafarm-api/internal/testutil" + "gorm.io/gorm" +) + +func TestCleanupPoolBatchDeletesDedicatedOwnerAndBatch(t *testing.T) { + db := testutil.Postgres(t) + admin := model.Admin{UniqueId: "pool-cleanup"} + if err := db.Create(&admin).Error; err != nil { + t.Fatal(err) + } + admitted, err := poolstore.Admit(context.Background(), db, poolstore.Admission{ + AdminID: admin.ID, Key: "cleanup", Digest: "cleanup", Label: "Cleanup", Count: 1, + Capacity: 1, + Session: model.SetupSession{ + Mode: model.ModeVtaOnly, Domain: "example.com", Subdomain: "vta-", + VtaImage: "example/vta:test", DidHostingServerURL: "https://dids.example.com", + }, + MemoryRequest: "32Mi", MemoryLimit: "64Mi", + }) + if err != nil { + t.Fatal(err) + } + batch := admitted.Batch + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + var session model.SetupSession + if err := db.First(&session, membership.SessionID).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&session).Update("status", model.StatusPoolWaiting).Error; err != nil { + t.Fatal(err) + } + if _, err := poolstore.QueueBatchCleanup(context.Background(), db, batch.ID); err != nil { + t.Fatal(err) + } + handler := &PoolHandler{db: db, setup: &SetupHandler{db: db}} + if err := handler.cleanupSession(context.Background(), session.ID); err != nil { + t.Fatal(err) + } + if err := db.First(&model.SetupSession{}, session.ID).Error; !errors.Is(err, gorm.ErrRecordNotFound) { + t.Fatalf("setup session lookup error = %v, want not found", err) + } + if err := db.First(&model.User{}, session.UserID).Error; !errors.Is(err, gorm.ErrRecordNotFound) { + t.Fatalf("dedicated user lookup error = %v, want not found", err) + } + if err := db.First(&model.PoolBatch{}, batch.ID).Error; !errors.Is(err, gorm.ErrRecordNotFound) { + t.Fatalf("pool batch lookup error = %v, want not found", err) + } +} + +func TestPoolBatchViewTreatsExpiredClaimAsLapsedWithoutWriting(t *testing.T) { + db := testutil.Postgres(t) + admin := model.Admin{UniqueId: "pool-read"} + if err := db.Create(&admin).Error; err != nil { + t.Fatal(err) + } + admitted, err := poolstore.Admit(context.Background(), db, poolstore.Admission{ + AdminID: admin.ID, Key: "read", Digest: "read", Label: "Read", Count: 1, Capacity: 1, + Session: model.SetupSession{ + Mode: model.ModeVtaOnly, Domain: "example.com", Subdomain: "vta-", + VtaImage: "example/vta:test", DidHostingServerURL: "https://dids.example.com", + }, + MemoryRequest: "32Mi", MemoryLimit: "64Mi", + }) + if err != nil { + t.Fatal(err) + } + session := admitted.Sessions[0] + if err := db.Model(&session).Update("status", model.StatusPoolWaiting).Error; err != nil { + t.Fatal(err) + } + past := time.Now().Add(-time.Minute) + claim := model.MobileConnection{ + ID: uuid.NewString(), SessionID: session.ID, VtaDid: "did:webvh:test", + Operation: "provision_vta", Status: "open", ExpiresAt: past, + } + if err := db.Create(&claim).Error; err != nil { + t.Fatal(err) + } + handler := &PoolHandler{db: db} + view, err := handler.batchView(context.Background(), admitted.Batch.ID, true) + if err != nil { + t.Fatal(err) + } + if len(view.Sessions) != 1 || view.Sessions[0].ClaimState != "lapsed" || !view.Sessions[0].Cleanable { + t.Fatalf("expired claim view = %+v", view.Sessions) + } + var stored model.MobileConnection + if err := db.First(&stored, "id = ?", claim.ID).Error; err != nil { + t.Fatal(err) + } + if stored.Status != "open" { + t.Fatalf("GET changed claim status to %q, want open", stored.Status) + } +} + +func TestPoolBatchViewDoesNotOfferCleanupWhilePreparing(t *testing.T) { + preparing := []string{ + model.StatusPoolCreating, + "vta_setup_running", + model.StatusPoolPublishing, + model.StatusPoolPrepared, + } + for _, status := range preparing { + view := buildPoolBatchView( + model.PoolBatch{ID: 1, RequestedCount: 1}, + []model.SetupSession{{ID: 1, UserID: 1, Status: status}}, + nil, + true, + ) + if len(view.Sessions) != 1 || view.Sessions[0].Cleanable { + t.Fatalf("status %q cleanable view = %+v, want false", status, view.Sessions) + } + } +} diff --git a/internal/handler/dashboard.go b/internal/handler/dashboard.go index f0cf70f..7fd3965 100644 --- a/internal/handler/dashboard.go +++ b/internal/handler/dashboard.go @@ -2,6 +2,7 @@ package handler import ( "context" + "errors" "net/http" "time" @@ -9,6 +10,7 @@ import ( "github.com/ic3software/vtafarm-api/internal/capacity" "github.com/ic3software/vtafarm-api/internal/k8s" + "github.com/ic3software/vtafarm-api/internal/model" "github.com/ic3software/vtafarm-api/internal/resourceprofile" "gorm.io/gorm" ) @@ -76,10 +78,13 @@ func (h *DashboardHandler) Get(c *gin.Context) { c.JSON(http.StatusBadGateway, gin.H{"error": "cluster stats: " + err.Error()}) return } - planningUsage := h.usage.Observe(time.Now(), stats.Nodes, stats.MetricsAvailable) + modes, calculated, err := h.capacityEstimates(ctx, stats) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "could not calculate capacity"}) + return + } nodes := make([]dashboardNode, len(stats.Nodes)) - var freeNodes []capacity.NodeFree var totalCPU, totalCPUReq, totalCPUUsed, totalMem, totalMemReq, totalMemUsed int64 for i, n := range stats.Nodes { nodes[i] = dashboardNode{ @@ -98,32 +103,9 @@ func (h *DashboardHandler) Get(c *gin.Context) { totalMem += n.MemAllocatableBytes totalMemReq += n.MemRequestedBytes totalMemUsed += n.MemUsedBytes - - // Estimates draw only on nodes new pods can land on. - if n.Schedulable { - usage, usageAvailable := planningUsage[n.Name] - var cpuFree, memFree int64 - if usageAvailable { - cpuFree = capacity.PlanningHeadroom( - n.CPUAllocatableMillis, - n.CPURequestedMillis, - usage.CPUMillis, - ) - memFree = capacity.PlanningHeadroom( - n.MemAllocatableBytes, - n.MemRequestedBytes, - usage.MemBytes, - ) - } - freeNodes = append(freeNodes, capacity.NodeFree{ - CPUMillis: cpuFree, - MemBytes: memFree, - }) - } } storageNodes := make([]dashboardStorageNode, len(stats.StorageNodes)) - var freeDisks []capacity.DiskFree var totalStorage, totalStorageReserved, totalStorageScheduled, totalStorageAvail, totalStorageSched int64 for i, s := range stats.StorageNodes { storageNodes[i] = dashboardStorageNode{ @@ -140,18 +122,12 @@ func (h *DashboardHandler) Get(c *gin.Context) { totalStorageScheduled += s.ScheduledBytes totalStorageAvail += s.AvailableBytes totalStorageSched += s.SchedulableBytes - freeDisks = append(freeDisks, capacity.DiskFree{Bytes: s.SchedulableBytes}) } replicas := stats.StorageReplicaCount estimates := gin.H{} - profiles, err := resourceprofile.Load(ctx, h.db) - if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": "could not load resource defaults"}) - return - } - for _, mode := range capacity.Modes(profiles) { - est := capacity.EstimateMode(mode, freeNodes, freeDisks, replicas, stats.StorageAvailable) + for _, mode := range modes { + est := calculated[mode.Name] var cpuCost, memCost, storageCost int64 for _, comp := range mode.Components { cpuCost += comp.CPUMillis @@ -199,3 +175,103 @@ func (h *DashboardHandler) Get(c *gin.Context) { "estimates": estimates, }) } + +// clusterCapacitySnapshot reads the slow cluster state without holding the pool +// mutation lock. Callers can load database-backed profiles concurrently. +func (h *DashboardHandler) clusterCapacitySnapshot(ctx context.Context) (*k8s.ClusterStats, error) { + if h.k8s == nil { + return nil, errors.New("kubernetes is not available") + } + return h.k8s.ClusterResourceStats(ctx) +} + +func (h *DashboardHandler) vtaOnlyCapacityFromSnapshot(stats *k8s.ClusterStats, profiles resourceprofile.Profiles) int { + _, estimates := h.capacityEstimatesWithProfiles(stats, profiles) + return estimates[capacity.VtaOnly.Name].Count +} + +func (h *DashboardHandler) capacityEstimates(ctx context.Context, stats *k8s.ClusterStats) ([]capacity.Mode, map[string]capacity.Estimate, error) { + profiles, err := resourceprofile.Load(ctx, h.db) + if err != nil { + return nil, nil, err + } + modes, estimates := h.capacityEstimatesWithProfiles(stats, profiles) + reserved, err := h.nonRunningPoolSessions(ctx) + if err != nil { + return nil, nil, err + } + if estimate, ok := estimates[capacity.VtaOnly.Name]; ok { + estimates[capacity.VtaOnly.Name] = subtractReservedSessions(estimate, reserved) + } + return modes, estimates, nil +} + +func (h *DashboardHandler) capacityEstimatesWithProfiles(stats *k8s.ClusterStats, profiles resourceprofile.Profiles) ([]capacity.Mode, map[string]capacity.Estimate) { + planningUsage := h.usage.Observe(time.Now(), stats.Nodes, stats.MetricsAvailable) + var freeNodes []capacity.NodeFree + for _, node := range stats.Nodes { + if !node.Schedulable { + continue + } + usage, ok := planningUsage[node.Name] + var cpuFree, memoryFree int64 + if ok { + cpuFree = capacity.PlanningHeadroom(node.CPUAllocatableMillis, node.CPURequestedMillis, usage.CPUMillis) + memoryFree = capacity.PlanningHeadroom(node.MemAllocatableBytes, node.MemRequestedBytes, usage.MemBytes) + } + freeNodes = append(freeNodes, capacity.NodeFree{CPUMillis: cpuFree, MemBytes: memoryFree}) + } + freeDisks := make([]capacity.DiskFree, 0, len(stats.StorageNodes)) + for _, disk := range stats.StorageNodes { + freeDisks = append(freeDisks, capacity.DiskFree{Bytes: disk.SchedulableBytes}) + } + modes := capacity.Modes(profiles) + estimates := make(map[string]capacity.Estimate, len(modes)) + for _, mode := range modes { + estimates[mode.Name] = capacity.EstimateMode( + mode, freeNodes, freeDisks, stats.StorageReplicaCount, stats.StorageAvailable, + ) + } + + return modes, estimates +} + +// A prebuilt, waiting Deployment has zero replicas and therefore does not appear in pod +// CPU or memory requests. Reserve one future VTA footprint until the pooled +// session is running; otherwise repeated pool admissions could all consume the +// same capacity displayed by the Dashboard. +func (h *DashboardHandler) nonRunningPoolSessions(ctx context.Context) (int, error) { + if !h.db.Migrator().HasTable(&model.PoolSession{}) { + return 0, nil + } + var count int64 + err := h.db.WithContext(ctx).Table("pool_sessions ps"). + Joins("JOIN setup_sessions s ON s.id = ps.session_id"). + Where("s.status <> ?", "running").Count(&count).Error + return int(count), err +} + +func subtractReservedSessions(estimate capacity.Estimate, reserved int) capacity.Estimate { + if reserved <= 0 { + return estimate + } + subtract := func(value int) int { + if value < 0 { + return value + } + return max(value-reserved, 0) + } + estimate.Count = subtract(estimate.Count) + estimate.ByCPU = subtract(estimate.ByCPU) + estimate.ByMemory = subtract(estimate.ByMemory) + estimate.ByStorage = subtract(estimate.ByStorage) + estimate.LimitingResource = "cpu" + minimum := estimate.ByCPU + if estimate.ByMemory < minimum { + estimate.LimitingResource, minimum = "memory", estimate.ByMemory + } + if estimate.ByStorage >= 0 && estimate.ByStorage < minimum { + estimate.LimitingResource = "storage" + } + return estimate +} diff --git a/internal/handler/dashboard_usage_test.go b/internal/handler/dashboard_usage_test.go index 589f52c..43cfab7 100644 --- a/internal/handler/dashboard_usage_test.go +++ b/internal/handler/dashboard_usage_test.go @@ -4,6 +4,7 @@ import ( "testing" "time" + "github.com/ic3software/vtafarm-api/internal/capacity" "github.com/ic3software/vtafarm-api/internal/k8s" ) @@ -42,6 +43,25 @@ func TestUsageHighWaterWithoutMetrics(t *testing.T) { } } +func TestSubtractReservedSessions(t *testing.T) { + got := subtractReservedSessions(capacity.Estimate{ + Count: 47, ByCPU: 112, ByMemory: 47, ByStorage: 173, LimitingResource: "memory", + }, 7) + if got.Count != 40 || got.ByCPU != 105 || got.ByMemory != 40 || got.ByStorage != 166 { + t.Fatalf("subtractReservedSessions() = %+v", got) + } + if got.LimitingResource != "memory" { + t.Fatalf("limiting resource = %q, want memory", got.LimitingResource) + } + + unknownStorage := subtractReservedSessions(capacity.Estimate{ + Count: 2, ByCPU: 2, ByMemory: 2, ByStorage: -1, + }, 5) + if unknownStorage.Count != 0 || unknownStorage.ByStorage != -1 { + t.Fatalf("unknown storage estimate = %+v", unknownStorage) + } +} + func assertNodeUsage(t *testing.T, got map[string]nodeUsage, node string, cpu, memory int64) { t.Helper() usage, ok := got[node] diff --git a/internal/handler/invitation.go b/internal/handler/invitation.go index 024a3f2..4f08c16 100644 --- a/internal/handler/invitation.go +++ b/internal/handler/invitation.go @@ -12,6 +12,7 @@ import ( "github.com/ic3software/vtafarm-api/internal/middleware" "github.com/ic3software/vtafarm-api/internal/model" + "github.com/ic3software/vtafarm-api/internal/publicid" ) const defaultInvitationTTL = 24 * time.Hour @@ -143,7 +144,10 @@ func (h *InvitationHandler) Register(c *gin.Context) { const maxAttempts = 5 var createErr error for range maxAttempts { - user.UniqueId = generateUniqueId() + user.UniqueId, createErr = publicid.NewAccount() + if createErr != nil { + break + } createErr = h.db.Create(&user).Error if createErr == nil { break diff --git a/internal/handler/setup.go b/internal/handler/setup.go index 2338b94..e6535a4 100644 --- a/internal/handler/setup.go +++ b/internal/handler/setup.go @@ -820,6 +820,17 @@ func (h *SetupHandler) Delete(c *gin.Context) { // user, AdminDeleteSession accepts any session. Everything after that point is // identical, so both funnel through here. func (h *SetupHandler) teardownSession(c *gin.Context, session *model.SetupSession) { + var poolMembership int64 + if h.db.Migrator().HasTable(&model.PoolSession{}) { + if err := h.db.Model(&model.PoolSession{}).Where("session_id = ?", session.ID).Count(&poolMembership).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to check pool membership"}) + return + } + } + if poolMembership != 0 && session.AdminDid == "" { + c.JSON(http.StatusConflict, gin.H{"error": "Use pool inventory cleanup for unclaimed VTAs. Confirmed claims are protected.", "reason": "pooled_session"}) + return + } if h.orch != nil { if err := h.orch.Cancel(session.ID); err != nil { c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Unable to stop provisioning before deletion. Please retry."}) @@ -848,6 +859,16 @@ func (h *SetupHandler) teardownSession(c *gin.Context, session *model.SetupSessi // response. The user/admin delete handlers and load-test batch cleanup share it // so all three remove the same DNS, DID-hosting, Kubernetes and Vault state. func (h *SetupHandler) teardownVtaOnlySession(ctx context.Context, session *model.SetupSession) (int, error) { + var poolMembership model.PoolSession + isPoolSession := false + if h.db.Migrator().HasTable(&model.PoolSession{}) { + result := h.db.Where("session_id = ?", session.ID).First(&poolMembership) + if result.Error != nil && !errors.Is(result.Error, gorm.ErrRecordNotFound) { + return http.StatusInternalServerError, errors.New("failed to check pool membership") + } + isPoolSession = result.Error == nil + } + if h.cf != nil && session.CFRecordID != "" { if err := h.cf.DeleteRecord(ctx, session.CFRecordID); err != nil { return http.StatusBadGateway, fmt.Errorf("failed to delete DNS record: %w", err) @@ -892,7 +913,49 @@ func (h *SetupHandler) teardownVtaOnlySession(ctx context.Context, session *mode h.orch.TeardownVaultSecrets(ctx, session.UserID, session.ID) } - if err := h.db.Delete(session).Error; err != nil { + if err := h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if isPoolSession { + if err := tx.Delete(&model.PoolSession{}, "session_id = ?", session.ID).Error; err != nil { + return err + } + } + if err := tx.Delete(session).Error; err != nil { + return err + } + if !isPoolSession { + return nil + } + var remainingSessions int64 + if err := tx.Model(&model.SetupSession{}).Where("user_id = ?", session.UserID).Count(&remainingSessions).Error; err != nil { + return err + } + if remainingSessions == 0 { + var user model.User + if err := tx.First(&user, session.UserID).Error; err != nil { + return err + } + var passkeys int64 + if err := tx.Model(&model.UserPasskey{}).Where("user_id = ?", user.ID).Count(&passkeys).Error; err != nil { + return err + } + if user.Email == nil && passkeys == 0 && !user.FullstackAccess { + if err := tx.Delete(&user).Error; err != nil { + return err + } + } + } + var remainingMembers int64 + if err := tx.Model(&model.PoolSession{}).Where("batch_id = ?", poolMembership.BatchID).Count(&remainingMembers).Error; err != nil { + return err + } + if remainingMembers == 0 { + if err := tx.Where("id = ? AND cleanup_requested_at IS NOT NULL", poolMembership.BatchID). + Delete(&model.PoolBatch{}).Error; err != nil { + return err + } + } + return nil + }); err != nil { return http.StatusInternalServerError, errors.New("failed to delete session") } @@ -910,6 +973,50 @@ func (h *SetupHandler) teardownVtaOnlySession(ctx context.Context, session *mode return http.StatusNoContent, nil } +// cleanupPoolSession finalizes a session that QueueCleanup already protected +// and moved to deleting. It is called by the durable pool runner and can be +// retried after a process restart. +func (h *SetupHandler) cleanupPoolSession(ctx context.Context, sessionID uint) error { + var session model.SetupSession + if err := h.db.WithContext(ctx).First(&session, sessionID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil + } + return err + } + var membership int64 + if err := h.db.WithContext(ctx).Model(&model.PoolSession{}). + Where("session_id = ?", sessionID).Count(&membership).Error; err != nil { + return err + } + if membership == 0 { + return nil + } + if session.Status != "deleting" || session.AdminDid != "" { + return errors.New("pool session is no longer eligible for cleanup") + } + var confirmed int64 + if err := h.db.WithContext(ctx).Model(&model.MobileConnection{}). + Where("session_id = ? AND confirmed_at IS NOT NULL", sessionID).Count(&confirmed).Error; err != nil { + return err + } + if confirmed != 0 { + return errors.New("confirmed pool claim cannot be cleaned as unused inventory") + } + if h.orch != nil { + if err := h.orch.Cancel(sessionID); err != nil { + return err + } + } + release, err := h.acquireSessionMaintenance(sessionID) + if err != nil { + return err + } + defer release() + _, err = h.teardownVtaOnlySession(ctx, &session) + return err +} + // GET /api/v1/setup/:id/logs func (h *SetupHandler) Logs(c *gin.Context) { publicID := c.Param("id") diff --git a/internal/handler/signup.go b/internal/handler/signup.go index b38ae3d..d7d9864 100644 --- a/internal/handler/signup.go +++ b/internal/handler/signup.go @@ -10,6 +10,7 @@ import ( "github.com/ic3software/vtafarm-api/internal/middleware" "github.com/ic3software/vtafarm-api/internal/model" + "github.com/ic3software/vtafarm-api/internal/publicid" ) type SignupHandler struct { @@ -60,7 +61,10 @@ func (h *SignupHandler) Signup(c *gin.Context) { const maxAttempts = 5 var createErr error for range maxAttempts { - user.UniqueId = generateUniqueId() + user.UniqueId, createErr = publicid.NewAccount() + if createErr != nil { + break + } createErr = h.db.Create(&user).Error if createErr == nil { break diff --git a/internal/handler/user.go b/internal/handler/user.go index 195aca7..f57f0a4 100644 --- a/internal/handler/user.go +++ b/internal/handler/user.go @@ -1,7 +1,6 @@ package handler import ( - "crypto/rand" "net/http" "strings" @@ -12,18 +11,6 @@ import ( "github.com/ic3software/vtafarm-api/internal/model" ) -const uniqueIdAlphabet = "abcdefghijklmnopqrstuvwxyz0123456789" - -func generateUniqueId() string { - b := make([]byte, 8) - _, _ = rand.Read(b) - result := make([]byte, 8) - for i, byt := range b { - result[i] = uniqueIdAlphabet[int(byt)%len(uniqueIdAlphabet)] - } - return string(result) -} - type UserHandler struct { db *gorm.DB } diff --git a/internal/k8s/cluster_stats.go b/internal/k8s/cluster_stats.go index 0545236..724e10f 100644 --- a/internal/k8s/cluster_stats.go +++ b/internal/k8s/cluster_stats.go @@ -6,6 +6,7 @@ import ( "fmt" "strconv" + "golang.org/x/sync/errgroup" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -74,17 +75,45 @@ type ClusterStats struct { // ClusterResourceStats reads node capacity, pod requests, live usage // (metrics-server), and Longhorn disk stats across every node in the cluster. func (c *Client) ClusterResourceStats(ctx context.Context) (*ClusterStats, error) { - nodeList, err := c.kube.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) - if err != nil { - return nil, fmt.Errorf("list nodes: %w", err) - } + var nodeList *corev1.NodeList + var podList *corev1.PodList + var cpuUsed, memUsed map[string]int64 + var metricsErr error + var storageNodes []StorageNodeStat + var dataWritten int64 + var replicas int + var storageErr error - // Non-terminal pods only: Succeeded/Failed pods release their requests. - podList, err := c.kube.CoreV1().Pods("").List(ctx, metav1.ListOptions{ - FieldSelector: "status.phase!=Succeeded,status.phase!=Failed", + group, groupCtx := errgroup.WithContext(ctx) + group.Go(func() error { + var err error + nodeList, err = c.kube.CoreV1().Nodes().List(groupCtx, metav1.ListOptions{}) + if err != nil { + return fmt.Errorf("list nodes: %w", err) + } + return nil }) - if err != nil { - return nil, fmt.Errorf("list pods: %w", err) + group.Go(func() error { + var err error + // Non-terminal pods only: Succeeded/Failed pods release their requests. + podList, err = c.kube.CoreV1().Pods("").List(groupCtx, metav1.ListOptions{ + FieldSelector: "status.phase!=Succeeded,status.phase!=Failed", + }) + if err != nil { + return fmt.Errorf("list pods: %w", err) + } + return nil + }) + group.Go(func() error { + cpuUsed, memUsed, metricsErr = c.nodeUsage(groupCtx) + return nil + }) + group.Go(func() error { + storageNodes, dataWritten, replicas, storageErr = c.longhornStats(groupCtx) + return nil + }) + if err := group.Wait(); err != nil { + return nil, err } cpuReq := map[string]int64{} @@ -101,7 +130,6 @@ func (c *Client) ClusterResourceStats(ctx context.Context) (*ClusterStats, error stats := &ClusterStats{StorageReplicaCount: 1} - cpuUsed, memUsed, metricsErr := c.nodeUsage(ctx) stats.MetricsAvailable = metricsErr == nil for i := range nodeList.Items { @@ -119,7 +147,6 @@ func (c *Client) ClusterResourceStats(ctx context.Context) (*ClusterStats, error stats.Nodes = append(stats.Nodes, stat) } - storageNodes, dataWritten, replicas, storageErr := c.longhornStats(ctx) if storageErr == nil { stats.StorageAvailable = true stats.StorageNodes = storageNodes diff --git a/internal/k8s/vta_resources.go b/internal/k8s/vta_resources.go index 1c77aad..647451f 100644 --- a/internal/k8s/vta_resources.go +++ b/internal/k8s/vta_resources.go @@ -44,9 +44,18 @@ func VtaTuningEnv() []corev1.EnvVar { // /work/vta to match, so VTA resolves its relative config.toml and data_dir // onto it. Port 8100. Idempotent — AlreadyExists is ignored. func (c *Client) CreateVtaDeployment(ctx context.Context, ns string, sessionID uint, image string, resources corev1.ResourceRequirements) error { + return c.createVtaDeployment(ctx, ns, sessionID, image, resources, 1) +} + +// CreateParkedVtaDeployment creates the final Deployment without starting a +// pod. Claim confirmation later uses the ordinary provisioning path to scale it. +func (c *Client) CreateParkedVtaDeployment(ctx context.Context, ns string, sessionID uint, image string, resources corev1.ResourceRequirements) error { + return c.createVtaDeployment(ctx, ns, sessionID, image, resources, 0) +} + +func (c *Client) createVtaDeployment(ctx context.Context, ns string, sessionID uint, image string, resources corev1.ResourceRequirements, replicas int32) error { name := vtaDeploymentName(sessionID) pvcName := VtaPVCName(sessionID) - replicas := int32(1) port := int32(8100) labels := map[string]string{ "app": "vta", diff --git a/internal/k8s/vta_resources_test.go b/internal/k8s/vta_resources_test.go index 03e6740..385c435 100644 --- a/internal/k8s/vta_resources_test.go +++ b/internal/k8s/vta_resources_test.go @@ -44,3 +44,21 @@ func TestCreateVtaDeploymentUsesHealthReadinessProbe(t *testing.T) { t.Fatalf("unexpected readiness timing: %+v", probe) } } + +func TestCreateParkedVtaDeploymentHasNoReplicas(t *testing.T) { + client := &Client{kube: fake.NewSimpleClientset()} + + if err := client.CreateParkedVtaDeployment(context.Background(), "test", 43, "example/vta:test", ComponentResources("10m", "32Mi", "128Mi")); err != nil { + t.Fatalf("CreateParkedVtaDeployment() error = %v", err) + } + + deployment, err := client.kube.AppsV1().Deployments("test").Get( + context.Background(), VtaDeploymentName(43), metav1.GetOptions{}, + ) + if err != nil { + t.Fatalf("get Deployment: %v", err) + } + if deployment.Spec.Replicas == nil || *deployment.Spec.Replicas != 0 { + t.Fatalf("replicas = %v, want 0", deployment.Spec.Replicas) + } +} diff --git a/internal/model/pool.go b/internal/model/pool.go new file mode 100644 index 0000000..1ad77f6 --- /dev/null +++ b/internal/model/pool.go @@ -0,0 +1,32 @@ +package model + +import "time" + +// Pool setup statuses extend the ordinary setup-session lifecycle while a VTA +// is being prepared as inactive inventory. +const ( + StatusPoolCreating = "pool_creating" + StatusPoolPublishing = "pool_publishing" + StatusPoolPrepared = "pool_prepared" + StatusPoolWaiting = "pool_waiting" +) + +// PoolBatch records one administrative request to create pooled VTAs. The +// actual users, sessions, images, infrastructure and lifecycle state remain in +// their existing tables. +type PoolBatch struct { + ID uint `gorm:"primaryKey" json:"id"` + CreatedBy uint `json:"created_by"` + Label string `json:"label"` + RequestedCount int `json:"requested_count"` + IdempotencyKey string `json:"-"` + RequestDigest string `json:"-"` + CleanupRequestedAt *time.Time `json:"-"` + CreatedAt time.Time `json:"created_at"` +} + +// PoolSession is membership, not another copy of setup-session state. +type PoolSession struct { + SessionID uint `gorm:"primaryKey" json:"session_id"` + BatchID uint `gorm:"not null;index" json:"batch_id"` +} diff --git a/internal/pool/runner.go b/internal/pool/runner.go new file mode 100644 index 0000000..a3d1f85 --- /dev/null +++ b/internal/pool/runner.go @@ -0,0 +1,163 @@ +package pool + +import ( + "context" + "database/sql/driver" + "errors" + "log" + "time" + + "github.com/ic3software/vtafarm-api/internal/model" + "gorm.io/gorm" +) + +const ( + workerCount = 4 + reconcileInterval = 10 * time.Second + workerLockClass = 565445 +) + +type Runner struct { + db *gorm.DB + prepare func(context.Context, uint) error + build func(context.Context, uint) error + cleanup func(context.Context, uint) error + queue chan uint +} + +func NewRunner( + db *gorm.DB, + prepare func(context.Context, uint) error, + build func(context.Context, uint) error, + cleanup func(context.Context, uint) error, +) *Runner { + return &Runner{db: db, prepare: prepare, build: build, cleanup: cleanup, queue: make(chan uint, 256)} +} + +func (r *Runner) Start(ctx context.Context) { + for range workerCount { + go r.runWorker(ctx) + } + go func() { + ticker := time.NewTicker(reconcileInterval) + defer ticker.Stop() + for { + r.reconcile(ctx) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + +func (r *Runner) Kick(sessionID uint) { + select { + case r.queue <- sessionID: + default: + // The periodic reconciliation loop is the durable fallback. + } +} + +func (r *Runner) reconcile(ctx context.Context) { + var sessionIDs []uint + err := r.db.WithContext(ctx).Table("pool_sessions ps"). + Select("ps.session_id"). + Joins("JOIN setup_sessions s ON s.id = ps.session_id"). + Where("s.status IN ?", []string{ + model.StatusPoolCreating, "vta_setup_running", model.StatusPoolPublishing, + model.StatusPoolPrepared, "deleting", + }).Order("ps.session_id").Limit(500).Scan(&sessionIDs).Error + if err != nil { + if ctx.Err() == nil { + log.Printf("[pool] reconcile failed: %v", err) + } + return + } + for _, sessionID := range sessionIDs { + r.Kick(sessionID) + } + if err := FinalizeEmptyBatchCleanups(ctx, r.db); err != nil && ctx.Err() == nil { + log.Printf("[pool] finalize empty batch cleanup failed: %v", err) + } +} + +func (r *Runner) runWorker(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case sessionID := <-r.queue: + r.runOne(ctx, sessionID) + } + } +} + +func (r *Runner) runOne(ctx context.Context, sessionID uint) { + pool, err := r.db.DB() + if err != nil { + return + } + conn, err := pool.Conn(ctx) + if err != nil { + return + } + defer conn.Close() + var locked bool + if err := conn.QueryRowContext(ctx, "SELECT pg_try_advisory_lock($1, $2)", workerLockClass, int64(sessionID)).Scan(&locked); err != nil || !locked { + return + } + defer func() { + releaseCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, err := conn.ExecContext(releaseCtx, "SELECT pg_advisory_unlock($1, $2)", workerLockClass, int64(sessionID)); err != nil { + _ = conn.Raw(func(any) error { return driver.ErrBadConn }) + } + }() + + var session model.SetupSession + if err := r.db.WithContext(ctx).First(&session, sessionID).Error; err != nil { + return + } + if session.Status == "deleting" { + if r.cleanup != nil { + if err := r.cleanup(ctx, sessionID); err != nil && ctx.Err() == nil { + r.recordCleanupError(sessionID, err) + } + } + return + } + if session.Status == model.StatusPoolWaiting || session.Status == "running" || session.Status == "failed" { + return + } + if r.prepare == nil || r.build == nil { + r.recordBuildError(sessionID, errors.New("pool worker is not configured")) + return + } + if err := r.prepare(ctx, sessionID); err != nil { + if ctx.Err() == nil { + r.recordBuildError(sessionID, err) + } + return + } + if err := r.build(ctx, sessionID); err != nil && ctx.Err() == nil { + r.recordBuildError(sessionID, err) + } +} + +func (r *Runner) recordBuildError(sessionID uint, err error) { + if updateErr := r.db.Model(&model.SetupSession{}). + Where("id = ? AND status <> 'deleting'", sessionID). + Updates(map[string]any{"status": "failed", "error_msg": err.Error()}).Error; updateErr != nil { + log.Printf("[pool] session %d: failed to record build error: %v", sessionID, updateErr) + } +} + +func (r *Runner) recordCleanupError(sessionID uint, err error) { + if updateErr := r.db.Model(&model.SetupSession{}). + Where("id = ? AND status = 'deleting'", sessionID). + Update("error_msg", err.Error()).Error; updateErr != nil { + log.Printf("[pool] session %d: failed to record cleanup error: %v", sessionID, updateErr) + } +} diff --git a/internal/pool/runner_test.go b/internal/pool/runner_test.go new file mode 100644 index 0000000..16b1f24 --- /dev/null +++ b/internal/pool/runner_test.go @@ -0,0 +1,49 @@ +package pool + +import ( + "context" + "testing" + + "github.com/ic3software/vtafarm-api/internal/model" +) + +func TestRunnerDispatchesDurablePoolWork(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 1, 1) + if err != nil { + t.Fatal(err) + } + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + prepared, built, cleaned := 0, 0, 0 + runner := NewRunner(db, + func(context.Context, uint) error { + prepared++ + return nil + }, + func(_ context.Context, sessionID uint) error { + built++ + return db.Model(&model.SetupSession{}).Where("id = ?", sessionID). + Update("status", model.StatusPoolWaiting).Error + }, + func(context.Context, uint) error { + cleaned++ + return nil + }, + ) + + runner.runOne(context.Background(), membership.SessionID) + if prepared != 1 || built != 1 || cleaned != 0 { + t.Fatalf("build dispatch counts = prepare:%d build:%d cleanup:%d", prepared, built, cleaned) + } + if err := db.Model(&model.SetupSession{}).Where("id = ?", membership.SessionID). + Update("status", "deleting").Error; err != nil { + t.Fatal(err) + } + runner.runOne(context.Background(), membership.SessionID) + if cleaned != 1 { + t.Fatalf("cleanup calls = %d, want 1", cleaned) + } +} diff --git a/internal/pool/store.go b/internal/pool/store.go new file mode 100644 index 0000000..8964cf8 --- /dev/null +++ b/internal/pool/store.go @@ -0,0 +1,577 @@ +package pool + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "sort" + "strings" + "time" + + "github.com/ic3software/vtafarm-api/internal/connection" + "github.com/ic3software/vtafarm-api/internal/model" + "github.com/ic3software/vtafarm-api/internal/publicid" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +var ( + ErrConflict = errors.New("pool state changed or session is protected") + ErrEmpty = errors.New("no waiting VTA is available in this batch") + ErrCapacity = errors.New("pool inventory limit reached") +) + +type CapacityError struct { + Available int +} + +func (e *CapacityError) Error() string { + return ErrCapacity.Error() +} + +func (e *CapacityError) Unwrap() error { + return ErrCapacity +} + +// All pool mutations use one transaction lock so two operators cannot select +// the same waiting session. +func lock(tx *gorm.DB) error { + return tx.Exec("SELECT pg_advisory_xact_lock(565443, 1)").Error +} + +type Admission struct { + AdminID uint + Key, Digest, Label string + Count int + Capacity int + Session model.SetupSession + MemoryRequest string + MemoryLimit string +} + +type AdmissionResult struct { + Batch *model.PoolBatch + Sessions []model.SetupSession + Created bool +} + +func LookupBatch(ctx context.Context, db *gorm.DB, adminID uint, key, digest string) (*model.PoolBatch, error) { + var batch model.PoolBatch + err := db.WithContext(ctx).Where("created_by = ? AND idempotency_key = ?", adminID, key).First(&batch).Error + if err == nil && batch.RequestDigest != digest { + return nil, ErrConflict + } + return &batch, err +} + +// Admit creates real users and setup sessions. pool_sessions only groups those +// sessions; it does not duplicate their ownership, infrastructure, or status. +func Admit(ctx context.Context, db *gorm.DB, input Admission) (*AdmissionResult, error) { + if input.Count < 1 || input.Key == "" || len(input.Key) > 128 || input.Session.VtaImage == "" { + return nil, errors.New("invalid pool admission") + } + result := &AdmissionResult{} + err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := lock(tx); err != nil { + return err + } + batch, created, err := admitBatchRow(ctx, tx, input) + if err != nil { + return err + } + if !created { + result.Batch = batch + return nil + } + users, err := createUsers(tx, input.Count) + if err != nil { + return err + } + sessions, err := createSetupSessions(tx, users, input.Session) + if err != nil { + return err + } + if err := createPoolMembershipsAndResources(tx, batch.ID, sessions, input.MemoryRequest, input.MemoryLimit); err != nil { + return err + } + result.Batch, result.Sessions, result.Created = batch, sessions, true + return nil + }) + return result, err +} + +func admitBatchRow(ctx context.Context, tx *gorm.DB, input Admission) (*model.PoolBatch, bool, error) { + var inserted []model.PoolBatch + err := tx.Raw(`WITH reserved AS ( + SELECT count(*)::integer AS count + FROM pool_sessions ps + JOIN setup_sessions s ON s.id = ps.session_id + WHERE s.status <> 'running' + ) + INSERT INTO pool_batches (created_by, label, requested_count, idempotency_key, request_digest) + SELECT ?, ?, ?, ?, ? FROM reserved + WHERE ? <= GREATEST(? - reserved.count, 0) + ON CONFLICT (created_by, idempotency_key) DO NOTHING + RETURNING *`, input.AdminID, input.Label, input.Count, input.Key, input.Digest, input.Count, input.Capacity). + Scan(&inserted).Error + if err != nil { + return nil, false, err + } + if len(inserted) == 1 { + return &inserted[0], true, nil + } + previous, err := LookupBatch(ctx, tx, input.AdminID, input.Key, input.Digest) + if err == nil { + return previous, false, nil + } + if !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, false, err + } + reserved, err := nonRunningSessionCount(tx) + if err != nil { + return nil, false, err + } + return nil, false, &CapacityError{Available: max(input.Capacity-reserved, 0)} +} + +func nonRunningSessionCount(tx *gorm.DB) (int, error) { + var count int64 + err := tx.Table("pool_sessions ps"). + Joins("JOIN setup_sessions s ON s.id = ps.session_id"). + Where("s.status <> ?", "running").Count(&count).Error + return int(count), err +} + +var newAccountID = publicid.NewAccount +var newVTAName = randomName + +func createUsers(tx *gorm.DB, count int) ([]model.User, error) { + users := make([]model.User, 0, count) + generated := make(map[string]struct{}, count) + for attempt := 0; len(users) < count && attempt < 5; attempt++ { + names, err := generateUniqueIDs(count-len(users), generated, newAccountID) + if err != nil { + return nil, err + } + placeholders := strings.TrimSuffix(strings.Repeat("(?),", len(names)), ",") + args := make([]any, len(names)) + for i := range names { + args[i] = names[i] + } + var inserted []model.User + if err := tx.Raw(`INSERT INTO users (unique_id) VALUES `+placeholders+` + ON CONFLICT (unique_id) DO NOTHING + RETURNING id, unique_id, created_at, updated_at`, args...).Scan(&inserted).Error; err != nil { + return nil, err + } + users = append(users, inserted...) + } + if len(users) != count { + return nil, errors.New("could not allocate user identifiers") + } + sort.Slice(users, func(i, j int) bool { return users[i].ID < users[j].ID }) + return users, nil +} + +func createSetupSessions(tx *gorm.DB, users []model.User, template model.SetupSession) ([]model.SetupSession, error) { + pending := make(map[uint]model.User, len(users)) + for i := range users { + pending[users[i].ID] = users[i] + } + sessions := make([]model.SetupSession, 0, len(users)) + generated := make(map[string]struct{}, len(users)) + for attempt := 0; len(pending) > 0 && attempt < 5; attempt++ { + pendingUsers := make([]model.User, 0, len(pending)) + for _, user := range pending { + pendingUsers = append(pendingUsers, user) + } + sort.Slice(pendingUsers, func(i, j int) bool { return pendingUsers[i].ID < pendingUsers[j].ID }) + names, err := generateUniqueIDs(len(pendingUsers), generated, newVTAName) + if err != nil { + return nil, err + } + candidates := make([]model.SetupSession, len(pendingUsers)) + userIDs := make([]uint, len(pendingUsers)) + for i := range pendingUsers { + session := template + session.ID, session.UserID, session.Status = 0, pendingUsers[i].ID, model.StatusPoolCreating + session.VtaName, session.AdminDid = names[i], "" + session.Subdomain = template.Subdomain + names[i] + session.VtaDidUrl = session.DidHostingServerURL + "/" + names[i] + "-vta" + session.CreatedAt, session.UpdatedAt = time.Time{}, time.Time{} + candidates[i], userIDs[i] = session, pendingUsers[i].ID + } + created := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&candidates) + if created.Error != nil { + return nil, created.Error + } + if created.RowsAffected == int64(len(candidates)) { + sessions = append(sessions, candidates...) + for i := range pendingUsers { + delete(pending, pendingUsers[i].ID) + } + continue + } + var inserted []model.SetupSession + if err := tx.Where("user_id IN ?", userIDs).Order("id").Find(&inserted).Error; err != nil { + return nil, err + } + for i := range inserted { + if _, ok := pending[inserted[i].UserID]; !ok { + continue + } + sessions = append(sessions, inserted[i]) + delete(pending, inserted[i].UserID) + } + } + if len(pending) != 0 { + return nil, errors.New("could not allocate VTA identifiers") + } + sort.Slice(sessions, func(i, j int) bool { return sessions[i].ID < sessions[j].ID }) + return sessions, nil +} + +func createPoolMembershipsAndResources(tx *gorm.DB, batchID uint, sessions []model.SetupSession, memoryRequest, memoryLimit string) error { + values := strings.TrimSuffix(strings.Repeat("(CAST(? AS bigint),CAST(? AS bigint)),", len(sessions)), ",") + args := make([]any, 0, len(sessions)*2+2) + for i := range sessions { + args = append(args, sessions[i].ID, batchID) + } + args = append(args, memoryRequest, memoryLimit) + result := tx.Exec(`WITH admitted(session_id, batch_id) AS (VALUES `+values+`), + memberships AS ( + INSERT INTO pool_sessions (session_id, batch_id) + SELECT session_id, batch_id FROM admitted + RETURNING session_id + ) + INSERT INTO workload_resources (setup_session_id, component, memory_request, memory_limit) + SELECT session_id, 'vta', ?, ? FROM memberships`, args...) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != int64(len(sessions)) { + return errors.New("pool membership bulk insert was incomplete") + } + return nil +} + +func generateUniqueIDs(count int, generated map[string]struct{}, next func() (string, error)) ([]string, error) { + values := make([]string, 0, count) + for draws := 0; len(values) < count && draws < count*20+100; draws++ { + value, err := next() + if err != nil { + return nil, err + } + if _, exists := generated[value]; exists { + continue + } + generated[value] = struct{}{} + values = append(values, value) + } + if len(values) != count { + return nil, errors.New("random identifier source produced too many duplicates") + } + return values, nil +} + +func randomName() (string, error) { + var value [6]byte + if _, err := rand.Read(value[:]); err != nil { + return "", err + } + return hex.EncodeToString(value[:]), nil +} + +// Reserve chooses one waiting session and creates its ordinary mobile claim. +// The QR lifecycle remains entirely in mobile_connections. +func Reserve(ctx context.Context, db *gorm.DB, batchID uint) (*model.MobileConnection, error) { + var result *model.MobileConnection + err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := lock(tx); err != nil { + return err + } + var candidates []model.SetupSession + if err := tx.Table("setup_sessions s"). + Select("s.*"). + Joins("JOIN pool_sessions ps ON ps.session_id = s.id"). + Where("ps.batch_id = ? AND s.status = ? AND s.admin_did = ''", batchID, model.StatusPoolWaiting). + Where(`NOT EXISTS ( + SELECT 1 FROM mobile_connections mc + WHERE mc.session_id = s.id + AND ( + (mc.status = 'open' AND mc.expires_at > clock_timestamp()) + OR (mc.status = 'claimed' AND mc.confirm_by > clock_timestamp()) + OR mc.status IN ('confirmed', 'provisioning') + ) + )`). + Order("s.id").Find(&candidates).Error; err != nil { + return err + } + for i := range candidates { + var session model.SetupSession + query := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).First(&session, candidates[i].ID) + if query.Error != nil { + if errors.Is(query.Error, gorm.ErrRecordNotFound) { + continue + } + return query.Error + } + if session.Status != model.StatusPoolWaiting || session.AdminDid != "" { + continue + } + mobile, err := connection.ChangePool(ctx, tx, session.ID) + if err != nil { + if errors.Is(err, connection.ErrConflict) || errors.Is(err, connection.ErrCooldown) { + continue + } + return err + } + result = mobile + return nil + } + return ErrEmpty + }) + return result, err +} + +func Confirm(ctx context.Context, db *gorm.DB, requestID string) (*model.MobileConnection, error) { + var result *model.MobileConnection + err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := lock(tx); err != nil { + return err + } + var candidate model.MobileConnection + if err := tx.Select("session_id").First(&candidate, "id = ?", requestID).Error; err != nil { + return err + } + if !IsSession(tx, candidate.SessionID) { + return ErrConflict + } + confirmed, err := connection.Confirm(ctx, tx, candidate.SessionID, requestID) + if err != nil { + var current model.MobileConnection + if loadErr := tx.First(¤t, "id = ?", requestID).Error; loadErr == nil && + (current.Status == "confirmed" || current.Status == "provisioning" || current.Status == "released") { + result = ¤t + return nil + } + return err + } + result, err = connection.StartProvisioning(ctx, tx, confirmed.ID) + return err + }) + return result, err +} + +func Reject(ctx context.Context, db *gorm.DB, requestID string) error { + return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := lock(tx); err != nil { + return err + } + var candidate model.MobileConnection + if err := tx.Select("session_id", "status").First(&candidate, "id = ?", requestID).Error; err != nil { + return err + } + if !IsSession(tx, candidate.SessionID) { + return ErrConflict + } + switch candidate.Status { + case "open": + _, err := connection.Change(ctx, tx, candidate.SessionID, "cancel", requestID) + return err + case "claimed": + _, err := connection.Reject(ctx, tx, candidate.SessionID, requestID) + return err + default: + return ErrConflict + } + }) +} + +func IsSession(db *gorm.DB, sessionID uint) bool { + var count int64 + return db.Model(&model.PoolSession{}).Where("session_id = ?", sessionID).Count(&count).Error == nil && count == 1 +} + +// RetryBuild returns an unclaimed failed member to the durable build queue. +func RetryBuild(ctx context.Context, db *gorm.DB, sessionID uint) error { + return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := lock(tx); err != nil { + return err + } + if err := tx.Exec("SELECT pg_advisory_xact_lock(?, ?)", workerLockClass, int64(sessionID)).Error; err != nil { + return err + } + if !IsSession(tx, sessionID) { + return ErrConflict + } + var session model.SetupSession + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&session, sessionID).Error; err != nil { + return err + } + if session.Status != "failed" || session.AdminDid != "" { + return ErrConflict + } + var protected int64 + if err := tx.Model(&model.MobileConnection{}). + Where("session_id = ? AND confirmed_at IS NOT NULL", sessionID). + Count(&protected).Error; err != nil { + return err + } + if protected != 0 { + return ErrConflict + } + return tx.Model(&session).Updates(map[string]any{ + "status": model.StatusPoolCreating, "error_msg": "", "failed_stage": "", + }).Error + }) +} + +// QueueCleanup accepts setup-session IDs. Confirmed claims remain protected +// even if activation failed and the setup session is no longer running. +func QueueCleanup(ctx context.Context, db *gorm.DB, sessionIDs []uint) error { + if len(sessionIDs) < 1 { + return ErrConflict + } + sessionIDs = append([]uint(nil), sessionIDs...) + sort.Slice(sessionIDs, func(i, j int) bool { return sessionIDs[i] < sessionIDs[j] }) + for i := 1; i < len(sessionIDs); i++ { + if sessionIDs[i] == sessionIDs[i-1] { + return ErrConflict + } + } + return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := lock(tx); err != nil { + return err + } + return queueCleanupBulk(tx, sessionIDs) + }) +} + +// QueueBatchCleanup atomically protects every member from new claims and marks +// it for durable teardown. The batch row is removed after its final member. +func QueueBatchCleanup(ctx context.Context, db *gorm.DB, batchID uint) ([]uint, error) { + var sessionIDs []uint + err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := lock(tx); err != nil { + return err + } + var batch model.PoolBatch + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&batch, batchID).Error; err != nil { + return err + } + if err := tx.Model(&model.PoolSession{}).Where("batch_id = ?", batchID). + Order("session_id").Pluck("session_id", &sessionIDs).Error; err != nil { + return err + } + if batch.CleanupRequestedAt != nil { + return nil + } + if len(sessionIDs) == 0 { + return tx.Delete(&batch).Error + } + if err := queueCleanupBulk(tx, sessionIDs); err != nil { + return err + } + return tx.Model(&batch).Update("cleanup_requested_at", gorm.Expr("clock_timestamp()")).Error + }) + return sessionIDs, err +} + +func queueCleanupBulk(tx *gorm.DB, sessionIDs []uint) error { + var memberships int64 + if err := tx.Model(&model.PoolSession{}). + Where("session_id IN ?", sessionIDs).Count(&memberships).Error; err != nil { + return err + } + if memberships != int64(len(sessionIDs)) { + return ErrConflict + } + + var sessions []model.SetupSession + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("id IN ?", sessionIDs).Order("id").Find(&sessions).Error; err != nil { + return err + } + if len(sessions) != len(sessionIDs) { + return ErrConflict + } + userIDs := make([]uint, len(sessions)) + for i := range sessions { + if sessions[i].AdminDid != "" || + (sessions[i].Status != model.StatusPoolWaiting && sessions[i].Status != "failed") { + return ErrConflict + } + userIDs[i] = sessions[i].UserID + } + + var users []model.User + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("id IN ?", userIDs).Order("id").Find(&users).Error; err != nil { + return err + } + if len(users) != len(userIDs) { + return ErrConflict + } + for i := range users { + if users[i].Email != nil || users[i].FullstackAccess { + return ErrConflict + } + } + var passkeys int64 + if err := tx.Model(&model.UserPasskey{}).Where("user_id IN ?", userIDs).Count(&passkeys).Error; err != nil { + return err + } + if passkeys != 0 { + return ErrConflict + } + + now, err := connection.Now(tx) + if err != nil { + return err + } + if err := tx.Model(&model.MobileConnection{}). + Where("session_id IN ? AND status = 'open' AND expires_at <= ?", sessionIDs, now). + Update("status", "lapsed").Error; err != nil { + return err + } + if err := tx.Model(&model.MobileConnection{}). + Where("session_id IN ? AND status = 'claimed' AND confirm_by <= ?", sessionIDs, now). + Updates(map[string]any{"status": "lapsed", "admin_did": ""}).Error; err != nil { + return err + } + var claims int64 + if err := tx.Model(&model.MobileConnection{}). + Where("session_id IN ? AND (confirmed_at IS NOT NULL OR status IN ?)", sessionIDs, []string{"open", "claimed", "confirmed", "provisioning", "released"}). + Count(&claims).Error; err != nil { + return err + } + if claims != 0 { + return ErrConflict + } + + updated := tx.Model(&model.SetupSession{}).Where("id IN ?", sessionIDs).Update("status", "deleting") + if updated.Error != nil { + return updated.Error + } + if updated.RowsAffected != int64(len(sessionIDs)) { + return ErrConflict + } + return nil +} + +func FinalizeBatchCleanup(ctx context.Context, db *gorm.DB, batchID uint) error { + return db.WithContext(ctx).Exec(` + DELETE FROM pool_batches b + WHERE b.id = ? + AND b.cleanup_requested_at IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM pool_sessions ps WHERE ps.batch_id = b.id)`, batchID).Error +} + +func FinalizeEmptyBatchCleanups(ctx context.Context, db *gorm.DB) error { + return db.WithContext(ctx).Exec(` + DELETE FROM pool_batches b + WHERE b.cleanup_requested_at IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM pool_sessions ps WHERE ps.batch_id = b.id)`).Error +} diff --git a/internal/pool/store_test.go b/internal/pool/store_test.go new file mode 100644 index 0000000..9871fbd --- /dev/null +++ b/internal/pool/store_test.go @@ -0,0 +1,647 @@ +package pool + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/ic3software/vtafarm-api/internal/connection" + "github.com/ic3software/vtafarm-api/internal/didkey" + "github.com/ic3software/vtafarm-api/internal/model" + "github.com/ic3software/vtafarm-api/internal/testutil" + "gorm.io/gorm" +) + +func poolDatabase(t *testing.T) *gorm.DB { + t.Helper() + return testutil.Postgres(t) +} + +func admission(t *testing.T, db *gorm.DB, count, available int) (*model.PoolBatch, bool, error) { + t.Helper() + admin := model.Admin{UniqueId: "pool-admin"} + if err := db.FirstOrCreate(&admin, "unique_id = ?", admin.UniqueId).Error; err != nil { + t.Fatal(err) + } + result, err := Admit(context.Background(), db, Admission{ + AdminID: admin.ID, + Key: "create-event-one", + Digest: "digest-one", + Label: "Event One", + Count: count, + Capacity: available, + Session: model.SetupSession{ + Mode: model.ModeVtaOnly, + Domain: "example.com", + Subdomain: "vta-", + VtaImage: "example/vta:test", + DidHostingServerURL: "https://dids.example.com", + DidHostingControlURL: "https://dids.example.com", + }, + MemoryRequest: "32Mi", + MemoryLimit: "64Mi", + }) + if err != nil { + return nil, false, err + } + return result.Batch, result.Created, nil +} + +func holdWorkerLock(t *testing.T, db *gorm.DB, sessionID uint) func() { + t.Helper() + sqlDB, err := db.DB() + if err != nil { + t.Fatal(err) + } + conn, err := sqlDB.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + if _, err := conn.ExecContext(context.Background(), "SELECT pg_advisory_lock($1, $2)", workerLockClass, int64(sessionID)); err != nil { + conn.Close() + t.Fatal(err) + } + var once sync.Once + release := func() { + once.Do(func() { + _, _ = conn.ExecContext(context.Background(), "SELECT pg_advisory_unlock($1, $2)", workerLockClass, int64(sessionID)) + _ = conn.Close() + }) + } + t.Cleanup(release) + return release +} + +func TestAdmitStoresOnlyBatchMembership(t *testing.T) { + db := poolDatabase(t) + batch, created, err := admission(t, db, 2, 2) + if err != nil || !created { + t.Fatalf("Admit() = %#v, %v, %v", batch, created, err) + } + + var memberships []model.PoolSession + if err := db.Where("batch_id = ?", batch.ID).Order("session_id").Find(&memberships).Error; err != nil { + t.Fatal(err) + } + if len(memberships) != 2 { + t.Fatalf("memberships = %d, want 2", len(memberships)) + } + for _, membership := range memberships { + var session model.SetupSession + if err := db.First(&session, membership.SessionID).Error; err != nil { + t.Fatal(err) + } + if session.Status != model.StatusPoolCreating || session.UserID == 0 { + t.Fatalf("session not admitted as a real owned VTA: %+v", session) + } + if len(session.VtaName) != 12 { + t.Fatalf("pool VTA name %q has length %d, want 12", session.VtaName, len(session.VtaName)) + } + var user model.User + if err := db.First(&user, session.UserID).Error; err != nil { + t.Fatal(err) + } + if len(user.UniqueId) != 8 { + t.Fatalf("pool user ID %q has length %d, want 8", user.UniqueId, len(user.UniqueId)) + } + } + + retried, created, err := admission(t, db, 2, 0) + if err != nil || created || retried.ID != batch.ID { + t.Fatalf("idempotent Admit() = %#v, %v, %v", retried, created, err) + } +} + +func TestAdmitUsesDashboardCapacity(t *testing.T) { + db := poolDatabase(t) + if _, _, err := admission(t, db, 48, 47); !errors.Is(err, ErrCapacity) { + t.Fatalf("Admit(48 with 47 available) error = %v, want ErrCapacity", err) + } + var batches int64 + if err := db.Model(&model.PoolBatch{}).Count(&batches).Error; err != nil { + t.Fatal(err) + } + if batches != 0 { + t.Fatalf("capacity rejection created %d batches", batches) + } + if _, created, err := admission(t, db, 47, 47); err != nil || !created { + t.Fatalf("Admit(47 with 47 available) created=%v error=%v", created, err) + } +} + +func TestAdmitBulkCreatesRequestedSizes(t *testing.T) { + for _, count := range []int{1, 20, 50} { + t.Run(fmt.Sprintf("count_%d", count), func(t *testing.T) { + db := poolDatabase(t) + started := time.Now() + batch, created, err := admission(t, db, count, count) + t.Logf("Admit(%d) completed in %s", count, time.Since(started)) + if err != nil || !created { + t.Fatalf("Admit(%d) created=%v error=%v", count, created, err) + } + var sessions, users, memberships, resources int64 + if err := db.Model(&model.PoolSession{}).Where("batch_id = ?", batch.ID).Count(&memberships).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}). + Where("id IN (SELECT session_id FROM pool_sessions WHERE batch_id = ?)", batch.ID).Count(&sessions).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.User{}). + Where("id IN (SELECT user_id FROM setup_sessions WHERE id IN (SELECT session_id FROM pool_sessions WHERE batch_id = ?))", batch.ID). + Count(&users).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.WorkloadResource{}). + Where("setup_session_id IN (SELECT session_id FROM pool_sessions WHERE batch_id = ?)", batch.ID).Count(&resources).Error; err != nil { + t.Fatal(err) + } + if sessions != int64(count) || users != int64(count) || memberships != int64(count) || resources != int64(count) { + t.Fatalf("rows sessions=%d users=%d memberships=%d resources=%d, want %d each", sessions, users, memberships, resources, count) + } + }) + } +} + +func TestAdmitBulkWriteFailureRollsBackEverything(t *testing.T) { + db := poolDatabase(t) + if err := db.Exec(` + CREATE FUNCTION reject_pool_workload() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + RAISE EXCEPTION 'injected workload failure'; + END $$; + CREATE TRIGGER reject_pool_workload BEFORE INSERT ON workload_resources + FOR EACH STATEMENT EXECUTE FUNCTION reject_pool_workload();`).Error; err != nil { + t.Fatal(err) + } + if _, _, err := admission(t, db, 20, 20); err == nil { + t.Fatal("Admit() error = nil, want injected bulk write failure") + } + for name, target := range map[string]any{ + "batches": &model.PoolBatch{}, "users": &model.User{}, "sessions": &model.SetupSession{}, + "memberships": &model.PoolSession{}, "resources": &model.WorkloadResource{}, + } { + var count int64 + if err := db.Model(target).Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatalf("rollback left %d %s", count, name) + } + } +} + +func TestAdmitConcurrentCapacityCannotBeExceeded(t *testing.T) { + db := poolDatabase(t) + admin := model.Admin{UniqueId: "pool-concur"} + if err := db.Create(&admin).Error; err != nil { + t.Fatal(err) + } + start := make(chan struct{}) + errs := make(chan error, 2) + for i := range 2 { + go func(index int) { + <-start + _, err := Admit(context.Background(), db, Admission{ + AdminID: admin.ID, Key: fmt.Sprintf("concurrent-%d", index), Digest: fmt.Sprintf("digest-%d", index), + Label: "Concurrent", Count: 2, Capacity: 3, + Session: model.SetupSession{ + Mode: model.ModeVtaOnly, Domain: "example.com", Subdomain: "vta-", + VtaImage: "example/vta:test", DidHostingServerURL: "https://dids.example.com", + }, + MemoryRequest: "32Mi", MemoryLimit: "64Mi", + }) + errs <- err + }(i) + } + close(start) + var accepted, rejected int + for range 2 { + err := <-errs + switch { + case err == nil: + accepted++ + case errors.Is(err, ErrCapacity): + rejected++ + default: + t.Fatalf("unexpected concurrent error: %v", err) + } + } + if accepted != 1 || rejected != 1 { + t.Fatalf("accepted=%d rejected=%d, want 1 each", accepted, rejected) + } + var memberships int64 + if err := db.Model(&model.PoolSession{}).Count(&memberships).Error; err != nil { + t.Fatal(err) + } + if memberships != 2 { + t.Fatalf("memberships = %d, want 2", memberships) + } +} + +func TestAdmitReplenishesIdentifierCollisions(t *testing.T) { + db := poolDatabase(t) + existingUser := model.User{UniqueId: "abcdefgh"} + if err := db.Create(&existingUser).Error; err != nil { + t.Fatal(err) + } + existingSession := testutil.SetupSession(t, db, model.ModeVtaOnly) + + accountIDs := []string{"abcdefgh", "ijklmnop", "qrstuvwx"} + vtaNames := []string{existingSession.VtaName, "aaaaaaaaaaaa", "bbbbbbbbbbbb"} + originalAccountID, originalVTAName := newAccountID, newVTAName + newAccountID = func() (string, error) { + value := accountIDs[0] + accountIDs = accountIDs[1:] + return value, nil + } + newVTAName = func() (string, error) { + value := vtaNames[0] + vtaNames = vtaNames[1:] + return value, nil + } + t.Cleanup(func() { + newAccountID, newVTAName = originalAccountID, originalVTAName + }) + + admin := model.Admin{UniqueId: "pool-collide"} + if err := db.Create(&admin).Error; err != nil { + t.Fatal(err) + } + result, err := Admit(context.Background(), db, Admission{ + AdminID: admin.ID, Key: "collision", Digest: "collision", Label: "Collision", Count: 2, Capacity: 2, + Session: model.SetupSession{ + Mode: model.ModeVtaOnly, Domain: "example.com", Subdomain: "vta-", + VtaImage: "example/vta:test", DidHostingServerURL: "https://dids.example.com", + }, + MemoryRequest: "32Mi", MemoryLimit: "64Mi", + }) + if err != nil { + t.Fatal(err) + } + if len(result.Sessions) != 2 { + t.Fatalf("returned sessions = %d, want 2", len(result.Sessions)) + } + for i := range result.Sessions { + if result.Sessions[i].VtaName == existingSession.VtaName { + t.Fatalf("reused colliding VTA name %q", existingSession.VtaName) + } + } +} + +func TestQueueBatchCleanupIsAtomicAndRetryable(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 2, 2) + if err != nil { + t.Fatal(err) + } + var memberships []model.PoolSession + if err := db.Where("batch_id = ?", batch.ID).Order("session_id").Find(&memberships).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}). + Where("id IN ?", []uint{memberships[0].SessionID, memberships[1].SessionID}). + Update("status", model.StatusPoolWaiting).Error; err != nil { + t.Fatal(err) + } + var protected model.SetupSession + if err := db.First(&protected, memberships[1].SessionID).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.User{}).Where("id = ?", protected.UserID). + Update("email", "claimed@example.com").Error; err != nil { + t.Fatal(err) + } + if _, err := QueueBatchCleanup(context.Background(), db, batch.ID); !errors.Is(err, ErrConflict) { + t.Fatalf("QueueBatchCleanup(protected) error = %v, want ErrConflict", err) + } + var deleting int64 + if err := db.Model(&model.SetupSession{}).Where("status = ?", "deleting").Count(&deleting).Error; err != nil { + t.Fatal(err) + } + if deleting != 0 { + t.Fatalf("atomic rejection left %d deleting sessions", deleting) + } + if err := db.Model(&model.User{}).Where("id = ?", protected.UserID). + Update("email", nil).Error; err != nil { + t.Fatal(err) + } + sessionIDs, err := QueueBatchCleanup(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + if len(sessionIDs) != 2 { + t.Fatalf("queued session IDs = %v, want 2", sessionIDs) + } + if err := db.First(batch, batch.ID).Error; err != nil { + t.Fatal(err) + } + if batch.CleanupRequestedAt == nil { + t.Fatal("cleanup_requested_at is nil") + } + retried, err := QueueBatchCleanup(context.Background(), db, batch.ID) + if err != nil || len(retried) != 2 { + t.Fatalf("QueueBatchCleanup(retry) = %v, %v", retried, err) + } +} + +func TestQueueBatchCleanupSkipsWorkerLockForWaitingSession(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 1, 1) + if err != nil { + t.Fatal(err) + } + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}).Where("id = ?", membership.SessionID). + Update("status", model.StatusPoolWaiting).Error; err != nil { + t.Fatal(err) + } + release := holdWorkerLock(t, db, membership.SessionID) + defer release() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if _, err := QueueBatchCleanup(ctx, db, batch.ID); err != nil { + t.Fatalf("QueueBatchCleanup(waiting) waited for worker lock: %v", err) + } +} + +func TestQueueBatchCleanupRejectsPreparingSessionWithoutWaitingForWorker(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 1, 1) + if err != nil { + t.Fatal(err) + } + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + release := holdWorkerLock(t, db, membership.SessionID) + defer release() + + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + if _, err := QueueBatchCleanup(ctx, db, batch.ID); !errors.Is(err, ErrConflict) { + t.Fatalf("QueueBatchCleanup(preparing) error = %v, want ErrConflict", err) + } + var session model.SetupSession + if err := db.First(&session, membership.SessionID).Error; err != nil { + t.Fatal(err) + } + if session.Status != model.StatusPoolCreating { + t.Fatalf("preparing session status = %q, want %q", session.Status, model.StatusPoolCreating) + } +} + +func TestQueueCleanupSkipsWorkerLockForWaitingSessions(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 2, 2) + if err != nil { + t.Fatal(err) + } + var memberships []model.PoolSession + if err := db.Where("batch_id = ?", batch.ID).Order("session_id").Find(&memberships).Error; err != nil { + t.Fatal(err) + } + sessionIDs := []uint{memberships[0].SessionID, memberships[1].SessionID} + if err := db.Model(&model.SetupSession{}).Where("id IN ?", sessionIDs). + Update("status", model.StatusPoolWaiting).Error; err != nil { + t.Fatal(err) + } + release := holdWorkerLock(t, db, sessionIDs[0]) + defer release() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := QueueCleanup(ctx, db, sessionIDs); err != nil { + t.Fatalf("QueueCleanup(waiting) waited for worker lock: %v", err) + } + var deleting int64 + if err := db.Model(&model.SetupSession{}). + Where("id IN ? AND status = ?", sessionIDs, "deleting").Count(&deleting).Error; err != nil { + t.Fatal(err) + } + if deleting != 2 { + t.Fatalf("deleting sessions = %d, want 2", deleting) + } +} + +func TestQueueCleanupRejectsNonPoolAndDuplicateSessions(t *testing.T) { + db := poolDatabase(t) + session := testutil.SetupSession(t, db, model.ModeVtaOnly) + if err := QueueCleanup(context.Background(), db, []uint{session.ID}); !errors.Is(err, ErrConflict) { + t.Fatalf("QueueCleanup(non-pool) error = %v, want ErrConflict", err) + } + var unchanged model.SetupSession + if err := db.First(&unchanged, session.ID).Error; err != nil { + t.Fatal(err) + } + if unchanged.Status != session.Status { + t.Fatalf("non-pool status = %q, want %q", unchanged.Status, session.Status) + } + + batch, _, err := admission(t, db, 1, 1) + if err != nil { + t.Fatal(err) + } + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + if err := QueueCleanup(context.Background(), db, []uint{membership.SessionID, membership.SessionID}); !errors.Is(err, ErrConflict) { + t.Fatalf("QueueCleanup(duplicate) error = %v, want ErrConflict", err) + } +} + +func TestConfirmedFailureCannotReturnToInventory(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 1, 1) + if err != nil { + t.Fatal(err) + } + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}).Where("id = ?", membership.SessionID).Updates(map[string]any{ + "status": model.StatusPoolWaiting, + "vta_did": "did:webvh:example.com:test", + }).Error; err != nil { + t.Fatal(err) + } + + reservation, err := Reserve(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + did, err := didkey.Generate() + if err != nil { + t.Fatal(err) + } + if _, err := connection.Claim(context.Background(), db, *reservation.ClaimHandle, did); err != nil { + t.Fatal(err) + } + if _, err := Confirm(context.Background(), db, reservation.ID); err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}).Where("id = ?", membership.SessionID).Update("status", "failed").Error; err != nil { + t.Fatal(err) + } + if err := QueueCleanup(context.Background(), db, []uint{membership.SessionID}); !errors.Is(err, ErrConflict) { + t.Fatalf("QueueCleanup() error = %v, want ErrConflict", err) + } +} + +func TestReserveDoesNotReuseActiveClaim(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 2, 2) + if err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}). + Where("id IN (SELECT session_id FROM pool_sessions WHERE batch_id = ?)", batch.ID). + Updates(map[string]any{"status": model.StatusPoolWaiting, "vta_did": "did:webvh:example.com:test"}).Error; err != nil { + t.Fatal(err) + } + first, err := Reserve(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + second, err := Reserve(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + if first.SessionID == second.SessionID { + t.Fatalf("two active claims used session %d", first.SessionID) + } +} + +func TestRejectCancelsOpenAndRejectsClaimedPoolRequests(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 2, 2) + if err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}). + Where("id IN (SELECT session_id FROM pool_sessions WHERE batch_id = ?)", batch.ID). + Updates(map[string]any{"status": model.StatusPoolWaiting, "vta_did": "did:webvh:example.com:test"}).Error; err != nil { + t.Fatal(err) + } + + openRequest, err := Reserve(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + if err := Reject(context.Background(), db, openRequest.ID); err != nil { + t.Fatalf("cancel open request: %v", err) + } + if err := db.First(openRequest, "id = ?", openRequest.ID).Error; err != nil { + t.Fatal(err) + } + if openRequest.Status != "lapsed" { + t.Fatalf("cancelled request status = %q, want lapsed", openRequest.Status) + } + + claimedRequest, err := Reserve(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + did, err := didkey.Generate() + if err != nil { + t.Fatal(err) + } + if _, err := connection.Claim(context.Background(), db, *claimedRequest.ClaimHandle, did); err != nil { + t.Fatal(err) + } + if err := Reject(context.Background(), db, claimedRequest.ID); err != nil { + t.Fatalf("reject claimed request: %v", err) + } + if err := db.First(claimedRequest, "id = ?", claimedRequest.ID).Error; err != nil { + t.Fatal(err) + } + if claimedRequest.Status != "rejected" { + t.Fatalf("rejected request status = %q, want rejected", claimedRequest.Status) + } +} + +func TestExpiredClaimCanBeReplacedAndCleaned(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 1, 1) + if err != nil { + t.Fatal(err) + } + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}).Where("id = ?", membership.SessionID). + Updates(map[string]any{"status": model.StatusPoolWaiting, "vta_did": "did:webvh:example.com:test"}).Error; err != nil { + t.Fatal(err) + } + first, err := Reserve(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + past := time.Now().Add(-10 * time.Minute) + if err := db.Model(first).Updates(map[string]any{"created_at": past, "expires_at": past}).Error; err != nil { + t.Fatal(err) + } + second, err := Reserve(context.Background(), db, batch.ID) + if err != nil { + t.Fatal(err) + } + if second.ID == first.ID || second.SessionID != first.SessionID { + t.Fatalf("replacement claim = %+v, first = %+v", second, first) + } + if err := db.Model(second).Updates(map[string]any{"created_at": past, "expires_at": past}).Error; err != nil { + t.Fatal(err) + } + if err := QueueCleanup(context.Background(), db, []uint{membership.SessionID}); err != nil { + t.Fatalf("QueueCleanup(expired claim) error = %v", err) + } + var session model.SetupSession + if err := db.First(&session, membership.SessionID).Error; err != nil { + t.Fatal(err) + } + if session.Status != "deleting" { + t.Fatalf("status = %q, want deleting", session.Status) + } +} + +func TestRetryBuildProtectsClaimedSession(t *testing.T) { + db := poolDatabase(t) + batch, _, err := admission(t, db, 1, 1) + if err != nil { + t.Fatal(err) + } + var membership model.PoolSession + if err := db.First(&membership, "batch_id = ?", batch.ID).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&model.SetupSession{}).Where("id = ?", membership.SessionID). + Updates(map[string]any{"status": "failed", "error_msg": "temporary"}).Error; err != nil { + t.Fatal(err) + } + if err := RetryBuild(context.Background(), db, membership.SessionID); err != nil { + t.Fatal(err) + } + var session model.SetupSession + if err := db.First(&session, membership.SessionID).Error; err != nil { + t.Fatal(err) + } + if session.Status != model.StatusPoolCreating || session.ErrorMsg != "" { + t.Fatalf("retried session = %+v", session) + } + if err := db.Model(&session).Updates(map[string]any{"status": "failed", "admin_did": "did:key:z6Mktest"}).Error; err != nil { + t.Fatal(err) + } + if err := RetryBuild(context.Background(), db, membership.SessionID); !errors.Is(err, ErrConflict) { + t.Fatalf("RetryBuild(claimed) error = %v, want ErrConflict", err) + } +} diff --git a/internal/publicid/user.go b/internal/publicid/user.go new file mode 100644 index 0000000..63fd4f2 --- /dev/null +++ b/internal/publicid/user.go @@ -0,0 +1,19 @@ +package publicid + +import "crypto/rand" + +const userAlphabet = "abcdefghijklmnopqrstuvwxyz0123456789" + +// NewAccount returns the public identifier format shared by admin, signup, +// invitation, and preallocated pool accounts. +func NewAccount() (string, error) { + random := make([]byte, 8) + if _, err := rand.Read(random); err != nil { + return "", err + } + result := make([]byte, len(random)) + for i, value := range random { + result[i] = userAlphabet[int(value)%len(userAlphabet)] + } + return string(result), nil +} diff --git a/internal/publicid/user_test.go b/internal/publicid/user_test.go new file mode 100644 index 0000000..cc8ddcc --- /dev/null +++ b/internal/publicid/user_test.go @@ -0,0 +1,23 @@ +package publicid + +import ( + "strings" + "testing" +) + +func TestNewAccountFormat(t *testing.T) { + for range 100 { + value, err := NewAccount() + if err != nil { + t.Fatal(err) + } + if len(value) != 8 { + t.Fatalf("len(%q) = %d, want 8", value, len(value)) + } + for _, char := range value { + if !strings.ContainsRune(userAlphabet, char) { + t.Fatalf("NewAccount() = %q, contains unexpected character %q", value, char) + } + } + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 3ea4589..99134d0 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -53,7 +53,7 @@ func Setup( r.Use(cors.New(cors.Config{ AllowOrigins: allowOrigins, AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"}, - AllowHeaders: []string{"Authorization", "Content-Type"}, + AllowHeaders: []string{"Authorization", "Content-Type", "Idempotency-Key"}, ExposeHeaders: []string{"Content-Length", "Content-Disposition"}, AllowCredentials: true, MaxAge: 12 * time.Hour, @@ -133,6 +133,10 @@ func Setup( // Image upgrades — shared by the admin batch routes and the user // self-service routes below; background runner, DB-backed queue. uph := handler.NewUpgradeHandler(db, upgradeRunner, ghcrClient, mediatorGhcrClient, didsGhcrClient, vtcGhcrClient) + mobile := handler.NewMobileConnectionHandler(db, orch, sh.KickAdditionalMobile, cfg.VTAClaim, trustTask.Enabled()) + dashH := handler.NewDashboardHandler(db, k8sClient) + poolH := handler.NewPoolHandler(db, sh, dashH, mobile) + poolH.Start(workerCtx) // Admin routes — cookie: vtafarm_admin adminAuth := v1.Group("", @@ -167,6 +171,17 @@ func Setup( adminAuth.GET("/admin/load-tests/:id", sh.AdminGetLoadTest) adminAuth.POST("/admin/load-tests/:id/check", sh.AdminCheckLoadTest) adminAuth.DELETE("/admin/load-tests/:id", sh.AdminDeleteLoadTest) + adminAuth.POST("/admin/pool-batches", poolH.CreateBatch) + adminAuth.GET("/admin/pool-batches", poolH.ListBatches) + adminAuth.GET("/admin/pool-batches/:id", poolH.GetBatch) + adminAuth.DELETE("/admin/pool-batches/:id", poolH.CleanupBatch) + adminAuth.POST("/admin/pool-batches/:id/claims", poolH.Reserve) + adminAuth.POST("/admin/pool-sessions/cleanup", poolH.Cleanup) + adminAuth.POST("/admin/pool-sessions/:id/retry", poolH.Retry) + poolClaims := adminAuth.Group("/admin/pool-claims", middleware.NoStore(), middleware.RateLimit(120, time.Minute)) + poolClaims.GET("/:request_id", poolH.GetClaim) + poolClaims.POST("/:request_id/confirm", poolH.ConfirmClaim) + poolClaims.POST("/:request_id/reject", poolH.RejectClaim) // Same teardown as the user-facing DELETE /setup/:id, but reaches any // user's session rather than only the caller's. Deleting the platform // stack additionally requires {"confirm": "