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/internal/api/serverless/client.go b/internal/api/serverless/client.go index 066111c..fc6598a 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. @@ -94,7 +95,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 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 }