diff --git a/docs/runware_serverless_apps.md b/docs/runware_serverless_apps.md index 3898fa5..df6ffda 100644 --- a/docs/runware_serverless_apps.md +++ b/docs/runware_serverless_apps.md @@ -41,6 +41,6 @@ runware serverless apps [flags] * [runware serverless apps stop](runware_serverless_apps_stop.md) - Stop a serverless application * [runware serverless apps tasks](runware_serverless_apps_tasks.md) - List and inspect application tasks * [runware serverless apps usage](runware_serverless_apps_usage.md) - Show usage for a serverless application -* [runware serverless apps versions](runware_serverless_apps_versions.md) - Inspect application versions +* [runware serverless apps versions](runware_serverless_apps_versions.md) - Manage application versions * [runware serverless apps workers](runware_serverless_apps_workers.md) - List workers for a serverless application diff --git a/docs/runware_serverless_apps_invoke.md b/docs/runware_serverless_apps_invoke.md index 2469160..3f5e081 100644 --- a/docs/runware_serverless_apps_invoke.md +++ b/docs/runware_serverless_apps_invoke.md @@ -16,6 +16,10 @@ to poll until the task is completed or failed. expires, the command polls the returned task id; it never treats expiry as a failure and never resubmits. +A client-generated task id is sent with every invoke. Omit --task-id to +generate one. Resubmitting the same id returns the task it already names +instead of starting a second run. + ``` runware serverless apps invoke [flags] ``` @@ -32,6 +36,9 @@ runware serverless apps invoke [flags] # async invoke and poll runware serverless apps invoke my-app infer --wait -f payload.json + + # retry a lost response without starting a second task + runware serverless apps invoke my-app infer --task-id 7c9e6679-7425-40de-944b-e07fc1f90ae7 -f payload.json ``` ### Options @@ -41,6 +48,7 @@ runware serverless apps invoke [flags] -h, --help help for invoke --poll-interval duration Polling interval when waiting for a task (default 2s) --sync Use sync invocation and wait for a terminal task + --task-id string Client task id (UUID); generated if omitted --wait Poll until the task is completed or failed ``` diff --git a/docs/runware_serverless_apps_versions.md b/docs/runware_serverless_apps_versions.md index 37712c1..d3a29d4 100644 --- a/docs/runware_serverless_apps_versions.md +++ b/docs/runware_serverless_apps_versions.md @@ -1,10 +1,10 @@ ## runware serverless apps versions -Inspect application versions +Manage application versions ### Synopsis -List and inspect immutable versions of a serverless application. +List, inspect, and activate immutable versions of a serverless application. ``` runware serverless apps versions [flags] @@ -28,6 +28,7 @@ runware serverless apps versions [flags] ### SEE ALSO * [runware serverless apps](runware_serverless_apps.md) - Manage deployed serverless applications +* [runware serverless apps versions activate](runware_serverless_apps_versions_activate.md) - Activate a ready application version * [runware serverless apps versions list](runware_serverless_apps_versions_list.md) - List versions of a serverless application * [runware serverless apps versions show](runware_serverless_apps_versions_show.md) - Show a version of a serverless application diff --git a/docs/runware_serverless_apps_versions_activate.md b/docs/runware_serverless_apps_versions_activate.md new file mode 100644 index 0000000..35e6038 --- /dev/null +++ b/docs/runware_serverless_apps_versions_activate.md @@ -0,0 +1,50 @@ +## runware serverless apps versions activate + +Activate a ready application version + +### Synopsis + +Activate a ready version by number, including rollback to an older version. + +The server accepts the deploy and returns immediately with the updated app. +Worker rollout is asynchronous; this command does not wait until workers are +healthy. Re-activating the currently active version is permitted and re-applies +it. On a stopped app the version is recorded and applied on resume. + +A missing app is 404. A missing version, a version that is not ready, or an +app that is deleting is 409. + +``` +runware serverless apps versions activate [flags] +``` + +### Examples + +``` + # list versions, then activate one + runware serverless apps versions list my-app + runware serverless apps versions activate my-app 2 + + # roll back to an older ready version + runware serverless apps versions activate my-app 1 +``` + +### Options + +``` + -h, --help help for activate +``` + +### Options inherited from parent commands + +``` + --debug Show full debug output + -F, --format string CLI output format: table, json, yaml (default "table") + --transport string Transport protocol: ws (WebSocket) or http (REST) (default "ws") + -v, --verbose Show request/response details +``` + +### SEE ALSO + +* [runware serverless apps versions](runware_serverless_apps_versions.md) - Manage application versions + diff --git a/docs/runware_serverless_apps_versions_list.md b/docs/runware_serverless_apps_versions_list.md index 1acb8e7..d0d010e 100644 --- a/docs/runware_serverless_apps_versions_list.md +++ b/docs/runware_serverless_apps_versions_list.md @@ -41,5 +41,5 @@ runware serverless apps versions list [flags] ### SEE ALSO -* [runware serverless apps versions](runware_serverless_apps_versions.md) - Inspect application versions +* [runware serverless apps versions](runware_serverless_apps_versions.md) - Manage application versions diff --git a/docs/runware_serverless_apps_versions_show.md b/docs/runware_serverless_apps_versions_show.md index bbcc90c..5e3bc47 100644 --- a/docs/runware_serverless_apps_versions_show.md +++ b/docs/runware_serverless_apps_versions_show.md @@ -34,5 +34,5 @@ runware serverless apps versions show [flags] ### SEE ALSO -* [runware serverless apps versions](runware_serverless_apps_versions.md) - Inspect application versions +* [runware serverless apps versions](runware_serverless_apps_versions.md) - Manage application versions diff --git a/internal/api/serverless/client.go b/internal/api/serverless/client.go index 066111c..8781e61 100644 --- a/internal/api/serverless/client.go +++ b/internal/api/serverless/client.go @@ -29,8 +29,9 @@ const defaultTimeout = 30 * time.Second const createAppTimeout = 5 * time.Minute // invokeSyncTimeout bounds startSyncTask. It must exceed the platform wait -// window so a 504/202 with taskId is received; a client-side timeout would -// lose the id and force a resubmit (a second billable run). +// window so a 202 with the accepted task is received; a client-side timeout +// would lose the response and force a resubmit (a second billable run unless +// the same client task id is reused). const invokeSyncTimeout = 5 * time.Minute // GpuType is the public catalogue entry for a supported GPU type. @@ -66,6 +67,9 @@ type AppUpdate = gen.AppUpdate // WorkerConfigPatch is a partial worker configuration for updateApp. type WorkerConfigPatch = gen.WorkerConfigPatch +// DeployRequest is the request body for DeployVersion. +type DeployRequest = gen.DeployRequest + // ListAppsParams are optional filters for ListApps. type ListAppsParams = gen.ListAppsParams @@ -94,7 +98,8 @@ type Task = gen.Task type TaskStatus = gen.TaskStatus // TaskPayload is the JSON object forwarded to an endpoint handler. -type TaskPayload = gen.TaskPayload +// It is the TaskInvocation.payload member, not the request body itself. +type TaskPayload = map[string]interface{} // ListTasksParams are optional filters for ListTasks. type ListTasksParams = gen.ListTasksParams @@ -447,6 +452,31 @@ func (c *Client) DeleteApp(ctx context.Context, appID string) (*App, error) { }) } +// DeployVersion activates a ready version by number. Worker rollout is +// asynchronous; the 202 App reflects the persisted intent (activeVersionId +// and status), not healthy workers. An older ready number is a rollback. +func (c *Client) DeployVersion(ctx context.Context, appID string, versionNumber int32) (*App, error) { + if c.apiKey == "" { + return nil, transport.ErrNoAPIKey + } + + resp, err := c.inner.DeployVersionWithResponse(ctx, appID, DeployRequest{ + VersionNumber: versionNumber, + }) + if err != nil { + return nil, fmt.Errorf("deploy version: %w", err) + } + + c.logResponse(ctx, resp.HTTPResponse, resp.Body) + + return acceptedApp("deploy version", resp.StatusCode(), resp.JSON202, resp.Body, lifecycleProblems{ + Unauthorized: resp.ApplicationproblemJSON401, + Forbidden: resp.ApplicationproblemJSON403, + NotFound: resp.ApplicationproblemJSON404, + Conflict: resp.ApplicationproblemJSON409, + }) +} + // lifecycleProblems are typed RFC 9457 bodies bound by the generated client. type lifecycleProblems struct { Unauthorized *gen.ProblemDetails diff --git a/internal/api/serverless/client_test.go b/internal/api/serverless/client_test.go index 9dfa639..f3d12c0 100644 --- a/internal/api/serverless/client_test.go +++ b/internal/api/serverless/client_test.go @@ -908,6 +908,109 @@ func TestGetVersion_NoAPIKey(t *testing.T) { } } +func TestDeployVersion(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + want := "/v1/apps/" + testAppID + "/deploy" + if r.Method != http.MethodPost || r.URL.Path != want { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read body: %v", err) + } + var req DeployRequest + if err := json.Unmarshal(body, &req); err != nil { + t.Errorf("body is not JSON: %s", body) + } + if req.VersionNumber != testVersionNumber { + t.Errorf("versionNumber = %d, want %d", req.VersionNumber, testVersionNumber) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{ + "appId":"my-app", + "appName":"My App", + "status":"initializing", + "activeVersionId":"` + testVersionID + `", + "configuration":{"maxWorkers":1,"idleTtlSecs":60,"scalingDelaySecs":10,"minWorkers":0,"gpusPerWorker":1,"concurrency":1,"computeType":"gpu"}, + "environmentVariables":[], + "secrets":[], + "createdAt":"2026-07-30T12:00:00Z", + "updatedAt":"2026-07-30T12:00:00Z" + }`)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + app, err := c.DeployVersion(context.Background(), testAppID, testVersionNumber) + if err != nil { + t.Fatalf("DeployVersion: %v", err) + } + if app.AppId != testAppID { + t.Fatalf("unexpected app: %+v", app) + } + if app.ActiveVersionId == nil || app.ActiveVersionId.String() != testVersionID { + t.Fatalf("unexpected activeVersionId: %+v", app.ActiveVersionId) + } +} + +func TestDeployVersion_Conflict(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusConflict) + _, _ = w.Write([]byte(`{"type":"about:blank","title":"Conflict","status":409,"detail":"Version is not ready"}`)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + _, err := c.DeployVersion(context.Background(), testAppID, testVersionNumber) + var re *transport.RunwareError + if !errors.As(err, &re) { + t.Fatalf("expected *transport.RunwareError, got %T: %v", err, err) + } + if re.Code != transport.CodeValidation { + t.Errorf("expected CodeValidation, got %v", re.Code) + } + if re.StatusCode != http.StatusConflict { + t.Errorf("expected status 409, got %d", re.StatusCode) + } + if re.Message != "Version is not ready" { + t.Errorf("unexpected message: %q", re.Message) + } +} + +func TestDeployVersion_NotFound(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"type":"about:blank","title":"Not Found","status":404,"detail":"No app 'missing' exists"}`)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + _, err := c.DeployVersion(context.Background(), "missing", testVersionNumber) + var re *transport.RunwareError + if !errors.As(err, &re) { + t.Fatalf("expected *transport.RunwareError, got %T: %v", err, err) + } + if re.Code != transport.CodeNotFound { + t.Errorf("expected CodeNotFound, got %v", re.Code) + } + if re.StatusCode != http.StatusNotFound { + t.Errorf("expected status 404, got %d", re.StatusCode) + } + if re.Message != "No app 'missing' exists" { + t.Errorf("unexpected message: %q", re.Message) + } +} + +func TestDeployVersion_NoAPIKey(t *testing.T) { + c := NewClient("", "https://example.invalid", slog.Default()) + if _, err := c.DeployVersion(context.Background(), testAppID, testVersionNumber); !errors.Is(err, transport.ErrNoAPIKey) { + t.Fatalf("expected ErrNoAPIKey, got %v", err) + } +} + func TestListWorkers(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { want := "/v1/apps/" + testAppID + "/workers" diff --git a/internal/api/serverless/tasks.go b/internal/api/serverless/tasks.go index 712d456..9c5e42f 100644 --- a/internal/api/serverless/tasks.go +++ b/internal/api/serverless/tasks.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/runware/runware-cli/internal/api/serverless/gen" "github.com/runware/runware-cli/internal/api/transport" ) @@ -44,18 +45,20 @@ func ValidateEndpointPath(path string) error { } // InvokeAsync starts a task and returns the accepted (typically pending) task. -func (c *Client) InvokeAsync(ctx context.Context, appID, endpointPath string, body TaskPayload) (*Task, error) { +// taskID is the client-owned identifier; an empty value is replaced with a new UUID. +func (c *Client) InvokeAsync(ctx context.Context, appID, endpointPath, taskID string, body TaskPayload) (*Task, error) { if c.apiKey == "" { return nil, transport.ErrNoAPIKey } if err := ValidateEndpointPath(endpointPath); err != nil { return nil, err } - if body == nil { - body = TaskPayload{} + invocation, err := newTaskInvocation(taskID, body) + if err != nil { + return nil, err } - resp, err := c.inner.StartAsyncTaskWithResponse(ctx, appID, endpointPath, gen.StartAsyncTaskJSONRequestBody(body)) + resp, err := c.inner.StartAsyncTaskWithResponse(ctx, appID, endpointPath, invocation) if err != nil { return nil, fmt.Errorf("invoke async: %w", err) } @@ -80,26 +83,29 @@ func (c *Client) InvokeAsync(ctx context.Context, appID, endpointPath string, bo return nil, problemToError(resp.ApplicationproblemJSON409, http.StatusConflict) case http.StatusUnprocessableEntity: return nil, problemToError(resp.ApplicationproblemJSON422, http.StatusUnprocessableEntity) + case http.StatusServiceUnavailable: + return nil, problemToError(resp.ApplicationproblemJSON503, http.StatusServiceUnavailable) default: return nil, problemFromBody(resp.Body, resp.StatusCode()) } } // InvokeSync starts a task and waits up to the platform wait window. -// A wait-window expiry (504 today, 202 after RUNSERV-547) is not a failure: -// the accepted task is returned so the caller can poll. Never resubmit. -func (c *Client) InvokeSync(ctx context.Context, appID, endpointPath string, body TaskPayload) (*Task, error) { +// A wait-window expiry is not a failure: the accepted task is returned so +// the caller can poll. Never resubmit. +func (c *Client) InvokeSync(ctx context.Context, appID, endpointPath, taskID string, body TaskPayload) (*Task, error) { if c.apiKey == "" { return nil, transport.ErrNoAPIKey } if err := ValidateEndpointPath(endpointPath); err != nil { return nil, err } - if body == nil { - body = TaskPayload{} + invocation, err := newTaskInvocation(taskID, body) + if err != nil { + return nil, err } - resp, err := c.innerWithMinTimeout(invokeSyncTimeout).StartSyncTaskWithResponse(ctx, appID, endpointPath, gen.StartSyncTaskJSONRequestBody(body)) + resp, err := c.innerWithMinTimeout(invokeSyncTimeout).StartSyncTaskWithResponse(ctx, appID, endpointPath, invocation) if err != nil { return nil, fmt.Errorf("invoke sync: %w", err) } @@ -113,20 +119,14 @@ func (c *Client) InvokeSync(ctx context.Context, appID, endpointPath string, bod } return resp.JSON200, nil case http.StatusAccepted: - // Settled shape (RUNSERV-547): wait expiry returns 202 + Task. + if resp.JSON202 != nil { + return resp.JSON202, nil + } task, err := taskFromBody(resp.Body) if err != nil { - if id := taskIDFromProblem(nil, resp.Body); id != "" { - return pendingTask(appID, id), nil - } return nil, fmt.Errorf("invoke sync: wait window expired without a task id") } return task, nil - case http.StatusGatewayTimeout: - if id := taskIDFromProblem(resp.ApplicationproblemJSON504, resp.Body); id != "" { - return pendingTask(appID, id), nil - } - return nil, fmt.Errorf("invoke sync: wait window expired without a task id") case http.StatusBadRequest: return nil, problemToError(resp.ApplicationproblemJSON400, http.StatusBadRequest) case http.StatusUnauthorized: @@ -139,6 +139,8 @@ func (c *Client) InvokeSync(ctx context.Context, appID, endpointPath string, bod return nil, problemToError(resp.ApplicationproblemJSON409, http.StatusConflict) case http.StatusUnprocessableEntity: return nil, problemToError(resp.ApplicationproblemJSON422, http.StatusUnprocessableEntity) + case http.StatusServiceUnavailable: + return nil, problemToError(resp.ApplicationproblemJSON503, http.StatusServiceUnavailable) default: return nil, problemFromBody(resp.Body, resp.StatusCode()) } @@ -150,7 +152,12 @@ func (c *Client) GetTask(ctx context.Context, appID, taskID string) (*Task, erro return nil, transport.ErrNoAPIKey } - resp, err := c.inner.GetTaskWithResponse(ctx, appID, taskID) + id, err := parseTaskID(taskID) + if err != nil { + return nil, err + } + + resp, err := c.inner.GetTaskWithResponse(ctx, appID, id) if err != nil { return nil, fmt.Errorf("get task: %w", err) } @@ -169,6 +176,8 @@ func (c *Client) GetTask(ctx context.Context, appID, taskID string) (*Task, erro return nil, problemToError(resp.ApplicationproblemJSON403, http.StatusForbidden) case http.StatusNotFound: return nil, problemToError(resp.ApplicationproblemJSON404, http.StatusNotFound) + case http.StatusServiceUnavailable: + return nil, problemToError(resp.ApplicationproblemJSON503, http.StatusServiceUnavailable) default: return nil, problemFromBody(resp.Body, resp.StatusCode()) } @@ -203,6 +212,8 @@ func (c *Client) ListTasks(ctx context.Context, appID string, params *ListTasksP return Page[Task]{}, problemToError(resp.ApplicationproblemJSON404, http.StatusNotFound) case http.StatusUnprocessableEntity: return Page[Task]{}, problemToError(resp.ApplicationproblemJSON422, http.StatusUnprocessableEntity) + case http.StatusServiceUnavailable: + return Page[Task]{}, problemToError(resp.ApplicationproblemJSON503, http.StatusServiceUnavailable) default: return Page[Task]{}, problemFromBody(resp.Body, resp.StatusCode()) } @@ -247,12 +258,33 @@ func (c *Client) WaitTask(ctx context.Context, appID, taskID string, interval ti } } -func pendingTask(appID, taskID string) *Task { - return &Task{ - Id: taskID, - AppId: appID, - Status: TaskStatusPending, +func newTaskInvocation(taskID string, body TaskPayload) (gen.TaskInvocation, error) { + id, err := resolveTaskID(taskID) + if err != nil { + return gen.TaskInvocation{}, err + } + if body == nil { + body = TaskPayload{} + } + return gen.TaskInvocation{ + Payload: body, + TaskId: id, + }, nil +} + +func resolveTaskID(taskID string) (uuid.UUID, error) { + if taskID == "" { + return uuid.New(), nil + } + return parseTaskID(taskID) +} + +func parseTaskID(taskID string) (uuid.UUID, error) { + id, err := uuid.Parse(taskID) + if err != nil { + return uuid.Nil, fmt.Errorf("invalid task id %q: must be a lowercase UUID", taskID) } + return id, nil } func taskFromBody(body []byte) (*Task, error) { @@ -266,23 +298,6 @@ func taskFromBody(body []byte) (*Task, error) { return &task, nil } -func taskIDFromProblem(p *gen.ProblemDetails, body []byte) string { - if p != nil && p.TaskId != nil && *p.TaskId != "" { - return *p.TaskId - } - if len(body) == 0 { - return "" - } - var parsed gen.ProblemDetails - if err := json.Unmarshal(body, &parsed); err != nil { - return "" - } - if parsed.TaskId == nil { - return "" - } - return *parsed.TaskId -} - func isNotFound(err error) bool { var re *transport.RunwareError return errors.As(err, &re) && re.Code == transport.CodeNotFound diff --git a/internal/api/serverless/tasks_test.go b/internal/api/serverless/tasks_test.go index 7ba756b..dc016e6 100644 --- a/internal/api/serverless/tasks_test.go +++ b/internal/api/serverless/tasks_test.go @@ -13,15 +13,16 @@ import ( "testing" "time" + "github.com/google/uuid" "github.com/runware/runware-cli/internal/api/transport" ) const ( - testTaskID = "task-1" + testTaskID = "7c9e6679-7425-40de-944b-e07fc1f90ae7" testEndpoint = "infer" - testTaskJSON = `{"id":"task-1","status":"pending","appId":"my-app","createdAt":"2026-07-30T12:00:00Z"}` - testTaskDone = `{"id":"task-1","status":"completed","appId":"my-app","createdAt":"2026-07-30T12:00:00Z","completedAt":"2026-07-30T12:00:05Z","output":{"ok":true}}` - testTaskFailed = `{"id":"task-1","status":"failed","appId":"my-app","createdAt":"2026-07-30T12:00:00Z","error":"oom killed"}` + testTaskJSON = `{"id":"7c9e6679-7425-40de-944b-e07fc1f90ae7","status":"pending","appId":"my-app","endpointPath":"infer","createdAt":"2026-07-30T12:00:00Z"}` + testTaskDone = `{"id":"7c9e6679-7425-40de-944b-e07fc1f90ae7","status":"completed","appId":"my-app","endpointPath":"infer","createdAt":"2026-07-30T12:00:00Z","completedAt":"2026-07-30T12:00:05Z","output":{"ok":true}}` + testTaskFailed = `{"id":"7c9e6679-7425-40de-944b-e07fc1f90ae7","status":"failed","appId":"my-app","endpointPath":"infer","createdAt":"2026-07-30T12:00:00Z","error":"oom killed"}` ) func TestValidateEndpointPath(t *testing.T) { @@ -70,10 +71,14 @@ func TestInvokeAsync(t *testing.T) { if err != nil { t.Errorf("read body: %v", err) } - var payload map[string]any - if err := json.Unmarshal(body, &payload); err != nil { + var invocation map[string]any + if err := json.Unmarshal(body, &invocation); err != nil { t.Errorf("body is not JSON: %s", body) } + if invocation["taskId"] != testTaskID { + t.Errorf("taskId = %v, want %s", invocation["taskId"], testTaskID) + } + payload, _ := invocation["payload"].(map[string]any) if payload["prompt"] != "hi" { t.Errorf("payload = %s", body) } @@ -84,7 +89,7 @@ func TestInvokeAsync(t *testing.T) { defer srv.Close() c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) - task, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, TaskPayload{"prompt": "hi"}) + task, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, testTaskID, TaskPayload{"prompt": "hi"}) if err != nil { t.Fatalf("InvokeAsync: %v", err) } @@ -98,7 +103,7 @@ func TestInvokeAsync(t *testing.T) { func TestInvokeAsync_RejectsLeadingSlash(t *testing.T) { c := NewClient("test-key", "https://example.invalid", slog.Default()) - _, err := c.InvokeAsync(context.Background(), testAppID, "/infer", nil) + _, err := c.InvokeAsync(context.Background(), testAppID, "/infer", "", nil) if err == nil || !strings.Contains(err.Error(), "bare segment") { t.Fatalf("expected leading-slash error, got %v", err) } @@ -106,7 +111,7 @@ func TestInvokeAsync_RejectsLeadingSlash(t *testing.T) { func TestInvokeAsync_NoAPIKey(t *testing.T) { c := NewClient("", "https://example.invalid", slog.Default()) - if _, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, nil); !errors.Is(err, transport.ErrNoAPIKey) { + if _, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, "", nil); !errors.Is(err, transport.ErrNoAPIKey) { t.Fatalf("expected ErrNoAPIKey, got %v", err) } } @@ -123,7 +128,7 @@ func TestInvokeSync_Completed(t *testing.T) { defer srv.Close() c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) - task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) + task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, "", nil) if err != nil { t.Fatalf("InvokeSync: %v", err) } @@ -132,30 +137,30 @@ func TestInvokeSync_Completed(t *testing.T) { } } -func TestInvokeSync_WaitExpiry504(t *testing.T) { - var posts atomic.Int32 +func TestInvokeAsync_GeneratesTaskID(t *testing.T) { + var gotID string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if strings.Contains(r.URL.Path, "/invoke-sync/") { - posts.Add(1) - w.Header().Set("Content-Type", "application/problem+json") - w.WriteHeader(http.StatusGatewayTimeout) - _, _ = w.Write([]byte(`{"type":"about:blank","title":"Gateway Timeout","status":504,"taskId":"task-1"}`)) - return + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read body: %v", err) } - t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + var invocation map[string]any + if err := json.Unmarshal(body, &invocation); err != nil { + t.Errorf("body is not JSON: %s", body) + } + gotID, _ = invocation["taskId"].(string) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(testTaskJSON)) })) defer srv.Close() c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) - task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) - if err != nil { - t.Fatalf("InvokeSync must not treat 504 as failure: %v", err) - } - if task.Id != testTaskID || task.Status != TaskStatusPending { - t.Fatalf("expected pending task %s, got %+v", testTaskID, task) + if _, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, "", nil); err != nil { + t.Fatalf("InvokeAsync: %v", err) } - if posts.Load() != 1 { - t.Fatalf("504 must not resubmit: got %d POSTs", posts.Load()) + if _, err := uuid.Parse(gotID); err != nil { + t.Fatalf("generated taskId %q is not a UUID: %v", gotID, err) } } @@ -168,7 +173,7 @@ func TestInvokeSync_WaitExpiry202(t *testing.T) { defer srv.Close() c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) - task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) + task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, "", nil) if err != nil { t.Fatalf("InvokeSync must not treat 202 as failure: %v", err) } @@ -177,21 +182,19 @@ func TestInvokeSync_WaitExpiry202(t *testing.T) { } } -func TestInvokeSync_504WithoutTaskID(t *testing.T) { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/problem+json") - w.WriteHeader(http.StatusGatewayTimeout) - _, _ = w.Write([]byte(`{"type":"about:blank","title":"Gateway Timeout","status":504}`)) - })) - defer srv.Close() - - c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) - _, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) - if err == nil || !strings.Contains(err.Error(), "without a task id") { - t.Fatalf("expected missing-id error, got %v", err) +func TestGetTask_InvalidID(t *testing.T) { + c := NewClient("test-key", "https://example.invalid", slog.Default()) + _, err := c.GetTask(context.Background(), testAppID, "not-a-uuid") + if err == nil || !strings.Contains(err.Error(), "invalid task id") { + t.Fatalf("expected invalid task id error, got %v", err) } - if strings.Contains(err.Error(), "timeout") || strings.Contains(err.Error(), "Timeout") { - t.Fatalf("must not report a timeout error: %v", err) +} + +func TestInvokeAsync_InvalidTaskID(t *testing.T) { + c := NewClient("test-key", "https://example.invalid", slog.Default()) + _, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, "not-a-uuid", nil) + if err == nil || !strings.Contains(err.Error(), "invalid task id") { + t.Fatalf("expected invalid task id error, got %v", err) } } diff --git a/internal/cmd/serverless/apps_invoke.go b/internal/cmd/serverless/apps_invoke.go index d163f57..c0d3664 100644 --- a/internal/cmd/serverless/apps_invoke.go +++ b/internal/cmd/serverless/apps_invoke.go @@ -20,6 +20,7 @@ func newAppsInvokeCmd(logger *log.Logger) *cobra.Command { sync bool wait bool bodyFile string + taskID string pollInterval time.Duration ) @@ -36,7 +37,11 @@ to poll until the task is completed or failed. --sync uses the sync invocation endpoint. If the platform wait window expires, the command polls the returned task id; it never treats expiry as -a failure and never resubmits.`, +a failure and never resubmits. + +A client-generated task id is sent with every invoke. Omit --task-id to +generate one. Resubmitting the same id returns the task it already names +instead of starting a second run.`, Example: ` # list endpoint paths, then invoke asynchronously runware serverless apps endpoints my-app runware serverless apps invoke my-app infer -f payload.json @@ -45,7 +50,10 @@ a failure and never resubmits.`, runware serverless apps invoke my-app infer --sync -f payload.json # async invoke and poll - runware serverless apps invoke my-app infer --wait -f payload.json`, + runware serverless apps invoke my-app infer --wait -f payload.json + + # retry a lost response without starting a second task + runware serverless apps invoke my-app infer --task-id 7c9e6679-7425-40de-944b-e07fc1f90ae7 -f payload.json`, Args: cobra.ExactArgs(2), RunE: func(cmd *cobra.Command, args []string) error { appID := args[0] @@ -60,9 +68,9 @@ a failure and never resubmits.`, var task *serverlessapi.Task if sync { - task, err = client.InvokeSync(cmd.Context(), appID, endpointPath, payload) + task, err = client.InvokeSync(cmd.Context(), appID, endpointPath, taskID, payload) } else { - task, err = client.InvokeAsync(cmd.Context(), appID, endpointPath, payload) + task, err = client.InvokeAsync(cmd.Context(), appID, endpointPath, taskID, payload) } if err != nil { spin.Stop() @@ -91,6 +99,7 @@ a failure and never resubmits.`, cmd.Flags().BoolVar(&sync, "sync", false, "Use sync invocation and wait for a terminal task") cmd.Flags().BoolVar(&wait, "wait", false, "Poll until the task is completed or failed") cmd.Flags().StringVarP(&bodyFile, "body", "f", "", "JSON payload file, or - for stdin (default {})") + cmd.Flags().StringVar(&taskID, "task-id", "", "Client task id (UUID); generated if omitted") cmd.Flags().DurationVar(&pollInterval, "poll-interval", 2*time.Second, "Polling interval when waiting for a task") return cmd } diff --git a/internal/cmd/serverless/apps_versions.go b/internal/cmd/serverless/apps_versions.go index ba53b51..8ff9bd5 100644 --- a/internal/cmd/serverless/apps_versions.go +++ b/internal/cmd/serverless/apps_versions.go @@ -14,11 +14,12 @@ import ( ) func newAppsVersionsCmd(logger *log.Logger) *cobra.Command { - cmd := stubGroup("versions", "Inspect application versions") - cmd.Long = "List and inspect immutable versions of a serverless application." + cmd := stubGroup("versions", "Manage application versions") + cmd.Long = "List, inspect, and activate immutable versions of a serverless application." cmd.AddCommand( newAppsVersionsListCmd(logger), newAppsVersionsShowCmd(logger), + newAppsVersionsActivateCmd(logger), ) return cmd } @@ -104,6 +105,49 @@ func newAppsVersionsShowCmd(logger *log.Logger) *cobra.Command { return cmd } +func newAppsVersionsActivateCmd(logger *log.Logger) *cobra.Command { + return &cobra.Command{ + Use: "activate ", + Short: "Activate a ready application version", + Long: `Activate a ready version by number, including rollback to an older version. + +The server accepts the deploy and returns immediately with the updated app. +Worker rollout is asynchronous; this command does not wait until workers are +healthy. Re-activating the currently active version is permitted and re-applies +it. On a stopped app the version is recorded and applied on resume. + +A missing app is 404. A missing version, a version that is not ready, or an +app that is deleting is 409.`, + Example: ` # list versions, then activate one + runware serverless apps versions list my-app + runware serverless apps versions activate my-app 2 + + # roll back to an older ready version + runware serverless apps versions activate my-app 1`, + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + appID := args[0] + n, err := parseVersionNumber(args[1]) + if err != nil { + return err + } + + spin := cmdutil.NewSpinner(fmt.Sprintf("Activating version %d on %s...", n, appID)) + spin.Start() + + client := serverlessapi.NewClient(config.GetAPIKey(), config.GetServerlessBaseURL(), slog.New(logger)) + app, err := client.DeployVersion(cmd.Context(), appID, n) + if err != nil { + spin.Stop() + return err + } + spin.Stop() + + return output.Print(cmdutil.FormatFor(cmd), appResult(*app)) + }, + } +} + func parseVersionNumber(s string) (int32, error) { n, err := strconv.ParseInt(s, 10, 32) if err != nil || n < 1 { diff --git a/internal/cmd/serverless/display.go b/internal/cmd/serverless/display.go index 9c64cdd..eaa9524 100644 --- a/internal/cmd/serverless/display.go +++ b/internal/cmd/serverless/display.go @@ -12,19 +12,20 @@ import ( ) const ( - colID = "ID" - colName = "Name" - colStatus = "Status" - colCreated = "Created" - colUpdated = "Updated" - colType = "Type" - colField = "Field" - colValue = "Value" - colApp = "App" - colKey = "Key" - colEnvVar = "Env var" - colError = "Error" - colCompleted = "Completed" + colID = "ID" + colName = "Name" + colStatus = "Status" + colActiveVersion = "Active version ID" + colCreated = "Created" + colUpdated = "Updated" + colType = "Type" + colField = "Field" + colValue = "Value" + colApp = "App" + colKey = "Key" + colEnvVar = "Env var" + colError = "Error" + colCompleted = "Completed" colComputeType = "Compute type" colGPUType = "GPU type" @@ -52,6 +53,7 @@ func (r appResult) Rows() [][]any { {colID, r.AppId}, {colName, r.AppName}, {colStatus, string(r.Status)}, + {colActiveVersion, formatOptionalUUID(r.ActiveVersionId)}, {colCreated, r.CreatedAt.Format(time.RFC3339)}, {colUpdated, r.UpdatedAt.Format(time.RFC3339)}, {colComputeType, string(cfg.ComputeType)}, diff --git a/internal/cmd/serverless/display_test.go b/internal/cmd/serverless/display_test.go index 2b1f97c..9b1dd00 100644 --- a/internal/cmd/serverless/display_test.go +++ b/internal/cmd/serverless/display_test.go @@ -159,12 +159,14 @@ func TestExtraStatusCursorFlag(t *testing.T) { func TestAppResult_IncludesConfiguration(t *testing.T) { gpu := testGPUType created := time.Date(2026, 7, 30, 12, 0, 0, 0, time.UTC) + activeVersion := uuid.MustParse("22222222-2222-2222-2222-222222222222") r := appResult{ - AppId: testAppID, - AppName: "My App", - Status: "active", - CreatedAt: created, - UpdatedAt: created, + AppId: testAppID, + AppName: "My App", + Status: "active", + ActiveVersionId: &activeVersion, + CreatedAt: created, + UpdatedAt: created, Configuration: serverlessapi.WorkerConfig{ ComputeType: "gpu", GpuType: &gpu, @@ -185,6 +187,7 @@ func TestAppResult_IncludesConfiguration(t *testing.T) { colID: testAppID, colName: "My App", colStatus: "active", + colActiveVersion: activeVersion.String(), colCreated: createdAt, colUpdated: createdAt, colComputeType: "gpu",