diff --git a/Makefile b/Makefile index 477d4d98..da5f4c4d 100644 --- a/Makefile +++ b/Makefile @@ -80,7 +80,7 @@ api-e2e-cloud: ## Run provisioning CLI API E2E tests against VOLCANO_API_URL and VOLCANO_API_E2E=1 go test ./tests/e2e/api -run '^TestAPIE2E(Smoke|Cloud)' -count=1 -timeout 240m localmode-e2e: ## Run destructive local-mode Docker smoke tests - VOLCANO_LOCALMODE_E2E=1 go test ./tests/e2e/localmode -run 'TestLocalModeE2E(Smoke|StopCleanSandboxes|DurableMessages)' -count=1 -timeout 35m + VOLCANO_LOCALMODE_E2E=1 go test ./tests/e2e/localmode -run 'TestLocalModeE2E(Smoke|StopCleanSandboxes|DurableMessages|DurableCallbacks)' -count=1 -timeout 35m lint: ## Run golangci-lint (includes gofmt, goimports, and go vet) go tool golangci-lint run ./... diff --git a/docs/README.md b/docs/README.md index 82dd9033..fbb39509 100644 --- a/docs/README.md +++ b/docs/README.md @@ -104,7 +104,7 @@ enable`. Turn it off on its own with `volcano feedback disable`. See | Project | create, list, get, rename, delete, select, get keys and usage | `projects …`, `use` | below | | Whole project | deploy variables, migrations, functions, frontends, and config in one run | `deploy`, `cloud deploy` | [deploy.md](deploy.md) | | Functions | deploy, invoke, inspect, set visibility, schedule, alias | `functions …` | [functions.md](functions.md) | -| Durable functions | deploy, start, inspect executions, read logs, schedule, decide approvals, send and read messages | `durable …`, `cloud durable …` | [durable-functions.md](durable-functions.md) | +| Durable functions | deploy, start, inspect executions, read logs, schedule, decide approvals, send and read messages, answer callbacks | `durable …`, `cloud durable …` | [durable-functions.md](durable-functions.md) | | Databases | create, inspect, delete, migrate | `databases …`, `migrations …` | [databases.md](databases.md) | | Storage | manage buckets, objects, policies | `storage …` | [storage.md](storage.md) | | Variables | deploy, list, get, delete | `variables …` | [variables.md](variables.md) | diff --git a/docs/durable-functions.md b/docs/durable-functions.md index c80e3f5e..42459e29 100644 --- a/docs/durable-functions.md +++ b/docs/durable-functions.md @@ -77,11 +77,14 @@ existing function's scope alone. | Send a message | `volcano cloud durable messages send [--data … \| --data-file …] [--idempotency-key …]` | | List messages | `volcano cloud durable messages list [--direction in\|out] [--name …] [--after …] [--limit 100] [--follow]` | | Open message waits | `volcano cloud durable messages waits ` | +| Succeed a callback | `volcano cloud durable callbacks succeed --result … \| --result-file …` | +| Fail a callback | `volcano cloud durable callbacks fail --type … [--message …]` | +| Heartbeat a callback | `volcano cloud durable callbacks heartbeat ` | Wherever a command takes a function, it takes the name or the id. `deploy` needs exactly one of `--all` and `-f`. `--yes` skips the confirmation prompt the three -destructive commands and the two approval decisions ask for. Every `approvals` -and `messages` command takes `--json`. Schedulers require `SUPERAGENT` and are capped at +destructive commands and the two approval decisions ask for. Every `approvals`, +`messages`, and `callbacks` command takes `--json`. Schedulers require `SUPERAGENT` and are capped at 5 per project across standard and durable functions together; a create beyond that answers `403`. @@ -315,6 +318,59 @@ project access token. A read-only token can list but not send. In local mode, drop `cloud`: `volcano durable messages send …` sends to an execution running on the local engine. +## Callbacks + +A workflow that calls `ctx.waitForCallback` hands a `vcb_` token to another +system and pauses until that system succeeds or fails the wait. See +[durable callbacks](/platform/functions/durable-callbacks) for the authoring +side. The `callbacks` commands are the token holder's side: + +```bash +# Still working: restart the wait's heartbeat timeout +volcano cloud durable callbacks heartbeat vcb_0000000000000000000000000000000000000000000 + +# Done: the workflow resumes with the result +volcano cloud durable callbacks succeed vcb_0000000000000000000000000000000000000000000 \ + --result '{"charge_id":"ch_1"}' + +# Or it failed: the workflow raises CallbackFailedError with this type and message +volcano cloud durable callbacks fail vcb_0000000000000000000000000000000000000000000 \ + --type CardDeclined --message "Insufficient funds" +``` + +```text +✓ Succeeded the callback; the execution resumes with the result +``` + +The token is the only credential these commands send, so they need no login +or project. Whoever holds the token can answer the wait, so treat it as a +secret. Pass `-` in place of the token to read it from stdin, which keeps it +out of shell history and the process list: + +```bash +printf '%s' "$CALLBACK_TOKEN" | volcano cloud durable callbacks succeed - --result-file charge.json +``` + +`--result` or `--result-file` is required: any JSON value other than `null`, up +to 64 KiB. `null` is refused because a wait that times out resolves with +`null`. `fail` requires `--type`, a name the workflow can branch on; +`--message` is optional. + +`succeed` and `fail` return once the execution has the outcome. When handing it +over takes longer, they report that Volcano is still delivering it. Volcano +finishes the delivery on its own, so do not send it again. + +A token takes one answer. A later `succeed`, `fail`, or `heartbeat` fails with +`the callback already has an outcome`, including a retry of the answer that went +through, so a holder retrying after a lost response can treat it as done. Once +the wait times out or its execution ends, every command fails and says so; +nothing will take an answer, so do not retry. A `heartbeat` that cannot reach +the execution right now leaves the wait unchanged; retry it with backoff. + +In local mode, drop `cloud`: `volcano durable callbacks succeed …` answers a +callback on the local engine. A callback's timeout and heartbeat timeout run in +real time locally. + ## Starting is asynchronous `start` returns as soon as the execution is accepted, with the execution's id diff --git a/docs/telemetry.md b/docs/telemetry.md index 58a464e1..c1fb411a 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -40,7 +40,7 @@ When a command fails, the event also carries the type of failure (`api`, code of an API error, and the error message. Before sending the message, the CLI replaces your home directory with `~`, strips terminal color codes, and removes bearer tokens, JWTs, Volcano keys and tokens (`pk-`, `pt-`, `sk-`, -`ak-`, `vbt_`, `vpg_`, `sbp1.`), passwords in URLs, email addresses, and URL +`ak-`, `vbt_`, `vpg_`, `vcb_`, `sbp1.`), passwords in URLs, email addresses, and URL query strings. It then cuts the message to 500 characters. Some errors can quote your files or data, so the CLI replaces that part with diff --git a/internal/api/durable_callbacks.go b/internal/api/durable_callbacks.go new file mode 100644 index 00000000..a31d8ded --- /dev/null +++ b/internal/api/durable_callbacks.go @@ -0,0 +1,84 @@ +package api + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + + "github.com/Kong/volcano-cli/internal/apiclient" +) + +// SucceedDurableCallback resumes the step waiting on token with result. The +// outcome is `succeeded` once the execution has it, or `delivering` when +// Volcano recorded it and is still handing it over. +// +// The token is the only credential the callback routes take, so c needs none. +func (c *Client) SucceedDurableCallback( + ctx context.Context, token string, result json.RawMessage, +) (*apiclient.DurableCallbackOutcome, error) { + body, err := callbackBody(apiclient.SucceedDurableCallbackJSONRequestBody{Token: token, Result: result}) + if err != nil { + return nil, err + } + resp, err := c.client.SucceedDurableCallbackWithBodyWithResponse(ctx, "application/json", body) + if err != nil { + return nil, err + } + if resp.JSON202 != nil { + return resp.JSON202, nil + } + return apiResult(resp.StatusCode(), resp.Body, resp.JSON200, + resp.JSON400, resp.JSON404, resp.JSON409, resp.JSON410, resp.JSON413, resp.JSON429) +} + +// FailDurableCallback resumes the step waiting on token by raising an error +// with errorType and message. An empty message is left out. +func (c *Client) FailDurableCallback( + ctx context.Context, token, errorType, message string, +) (*apiclient.DurableCallbackOutcome, error) { + failure := apiclient.DurableCallbackError{Type: errorType} + if message != "" { + failure.Message = &message + } + body, err := callbackBody(apiclient.FailDurableCallbackJSONRequestBody{Token: token, Error: failure}) + if err != nil { + return nil, err + } + resp, err := c.client.FailDurableCallbackWithBodyWithResponse(ctx, "application/json", body) + if err != nil { + return nil, err + } + if resp.JSON202 != nil { + return resp.JSON202, nil + } + return apiResult(resp.StatusCode(), resp.Body, resp.JSON200, + resp.JSON400, resp.JSON404, resp.JSON409, resp.JSON410, resp.JSON413, resp.JSON429) +} + +// HeartbeatDurableCallback tells the step waiting on token that the work is +// still going, restarting its heartbeat timeout. +func (c *Client) HeartbeatDurableCallback(ctx context.Context, token string) (*apiclient.DurableCallbackOutcome, error) { + resp, err := c.client.HeartbeatDurableCallbackWithResponse(ctx, apiclient.HeartbeatDurableCallbackJSONRequestBody{ + Token: token, + }) + if err != nil { + return nil, err + } + return apiResult(resp.StatusCode(), resp.Body, resp.JSON200, + resp.JSON400, resp.JSON404, resp.JSON409, resp.JSON410, resp.JSON413, resp.JSON429, resp.JSON503) +} + +// callbackBody encodes an answer without HTML escaping. Escaping turns each +// <, > or & into six bytes, so a result or message well inside its limit could +// otherwise push the request past the route's body limit. +func callbackBody(answer any) (io.Reader, error) { + var body bytes.Buffer + encoder := json.NewEncoder(&body) + encoder.SetEscapeHTML(false) + if err := encoder.Encode(answer); err != nil { + return nil, fmt.Errorf("failed to encode the callback answer: %w", err) + } + return &body, nil +} diff --git a/internal/api/durable_callbacks_test.go b/internal/api/durable_callbacks_test.go new file mode 100644 index 00000000..daed381c --- /dev/null +++ b/internal/api/durable_callbacks_test.go @@ -0,0 +1,170 @@ +package api + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Kong/volcano-cli/internal/apiclient" +) + +const durableTestCallbackToken = "vcb_0000000000000000000000000000000000000000000" + +// Each answer posts the token in the body to its own route, and a 202 is an +// outcome still being delivered rather than an error. +func TestDurableCallbackWireShape(t *testing.T) { + for _, tc := range []struct { + name string + path string + status int + answer func(*Client) (*apiclient.DurableCallbackOutcome, error) + wantBody string + wantStatus apiclient.DurableCallbackStatus + }{ + { + name: "succeed", path: "/durable-callbacks/succeed", status: http.StatusOK, + answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) { + return c.SucceedDurableCallback(context.Background(), durableTestCallbackToken, + json.RawMessage(`{"amount":12345678901234567890}`)) + }, + wantBody: `{"token":"` + durableTestCallbackToken + `","result":{"amount":12345678901234567890}}`, + wantStatus: apiclient.DurableCallbackStatusSucceeded, + }, + { + name: "succeed still delivering", path: "/durable-callbacks/succeed", status: http.StatusAccepted, + answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) { + return c.SucceedDurableCallback(context.Background(), durableTestCallbackToken, json.RawMessage(`0`)) + }, + wantBody: `{"token":"` + durableTestCallbackToken + `","result":0}`, + wantStatus: apiclient.DurableCallbackStatusDelivering, + }, + { + name: "fail with a message", path: "/durable-callbacks/fail", status: http.StatusOK, + answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) { + return c.FailDurableCallback(context.Background(), durableTestCallbackToken, + "CardDeclined", "Insufficient funds") + }, + wantBody: `{"token":"` + durableTestCallbackToken + + `","error":{"type":"CardDeclined","message":"Insufficient funds"}}`, + wantStatus: apiclient.DurableCallbackStatusFailed, + }, + { + name: "fail still delivering, without a message", path: "/durable-callbacks/fail", status: http.StatusAccepted, + answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) { + return c.FailDurableCallback(context.Background(), durableTestCallbackToken, "CardDeclined", "") + }, + wantBody: `{"token":"` + durableTestCallbackToken + `","error":{"type":"CardDeclined"}}`, + wantStatus: apiclient.DurableCallbackStatusDelivering, + }, + { + name: "heartbeat", path: "/durable-callbacks/heartbeat", status: http.StatusOK, + answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) { + return c.HeartbeatDurableCallback(context.Background(), durableTestCallbackToken) + }, + wantBody: `{"token":"` + durableTestCallbackToken + `"}`, + wantStatus: apiclient.DurableCallbackStatusWaiting, + }, + } { + t.Run(tc.name, func(t *testing.T) { + var body []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodPost, r.Method) + assert.Equal(t, tc.path, r.URL.Path) + assert.Empty(t, r.Header.Get("Authorization"), "the token is the only credential") + var err error + body, err = io.ReadAll(r.Body) + assert.NoError(t, err) + writeAPIJSON(t, w, tc.status, map[string]any{"status": tc.wantStatus}) + })) + defer server.Close() + + client, err := NewClient(server.URL, "", WithHTTPClient(server.Client())) + require.NoError(t, err) + + outcome, err := tc.answer(client) + require.NoError(t, err) + assert.Equal(t, tc.wantStatus, outcome.Status) + assert.JSONEq(t, tc.wantBody, string(body)) + }) + } +} + +// An answer is sent without HTML escaping, which would grow each <, > and & to +// six bytes and could push an answer inside its limit past the body limit. +func TestDurableCallbackAnswersAreNotHTMLEscaped(t *testing.T) { + for _, tc := range []struct { + name string + answer func(*Client) (*apiclient.DurableCallbackOutcome, error) + want string + }{ + { + name: "succeed", + answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) { + return c.SucceedDurableCallback(context.Background(), durableTestCallbackToken, + json.RawMessage(`{"note":"A & B"}`)) + }, + want: `"note":"A & B"`, + }, + { + name: "fail", + answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) { + return c.FailDurableCallback(context.Background(), durableTestCallbackToken, + "Rejected", "A & B") + }, + want: `"message":"A & B"`, + }, + } { + t.Run(tc.name, func(t *testing.T) { + var body []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "application/json", r.Header.Get("Content-Type")) + var err error + body, err = io.ReadAll(r.Body) + assert.NoError(t, err) + writeAPIJSON(t, w, http.StatusOK, map[string]any{"status": "succeeded"}) + })) + defer server.Close() + + client, err := NewClient(server.URL, "", WithHTTPClient(server.Client())) + require.NoError(t, err) + + _, err = tc.answer(client) + require.NoError(t, err) + assert.Contains(t, string(body), tc.want) + }) + } +} + +// The refusals that share a status are told apart by their code. +func TestDurableCallbackSurfacesTheRefusalCode(t *testing.T) { + for _, tc := range []struct { + status int + code string + }{ + {status: http.StatusNotFound, code: "callback_not_found"}, + {status: http.StatusConflict, code: "callback_completed"}, + {status: http.StatusGone, code: "callback_expired"}, + {status: http.StatusGone, code: "execution_ended"}, + } { + t.Run(tc.code, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeAPIJSON(t, w, tc.status, map[string]any{"error": "refused", "code": tc.code}) + })) + defer server.Close() + + client, err := NewClient(server.URL, "", WithHTTPClient(server.Client())) + require.NoError(t, err) + + _, err = client.HeartbeatDurableCallback(context.Background(), durableTestCallbackToken) + require.Error(t, err) + assert.Equal(t, tc.status, Status(err)) + assert.Equal(t, tc.code, Code(err)) + }) + } +} diff --git a/internal/apiclient/client.gen.go b/internal/apiclient/client.gen.go index aa1ea12a..ba357890 100644 --- a/internal/apiclient/client.gen.go +++ b/internal/apiclient/client.gen.go @@ -894,6 +894,30 @@ func (e DurableApprovalStatus) Valid() bool { } } +// Defines values for DurableCallbackStatus. +const ( + DurableCallbackStatusDelivering DurableCallbackStatus = "delivering" + DurableCallbackStatusFailed DurableCallbackStatus = "failed" + DurableCallbackStatusSucceeded DurableCallbackStatus = "succeeded" + DurableCallbackStatusWaiting DurableCallbackStatus = "waiting" +) + +// Valid indicates whether the value is a known member of the DurableCallbackStatus enum. +func (e DurableCallbackStatus) Valid() bool { + switch e { + case DurableCallbackStatusDelivering: + return true + case DurableCallbackStatusFailed: + return true + case DurableCallbackStatusSucceeded: + return true + case DurableCallbackStatusWaiting: + return true + default: + return false + } +} + // Defines values for DurableExecutionOperationKind. const ( DurableExecutionOperationKindCallback DurableExecutionOperationKind = "callback" @@ -6719,6 +6743,38 @@ type DurableApprovalStats struct { // is final. type DurableApprovalStatus string +// DurableCallbackError The error the waiting step raises. +type DurableCallbackError struct { + // Message A human-readable explanation of up to 32,768 bytes of UTF-8, with + // tab and newline as its only control characters. A character + // outside ASCII takes more than one byte, so such text reaches the + // limit before `maxLength`. + Message *string `json:"message,omitempty"` + + // Type A name workflows can branch on, such as `PaymentDeclined`. Up to + // 256 characters, without control characters. + Type string `json:"type"` +} + +// DurableCallbackOutcome defines model for DurableCallbackOutcome. +type DurableCallbackOutcome struct { + // Status `waiting` until the token is succeeded or failed. `delivering` while + // the outcome is on its way to the workflow; Volcano finishes the + // delivery on its own. `succeeded` or `failed` once the workflow has the + // outcome. + Status DurableCallbackStatus `json:"status"` +} + +// DurableCallbackStatus `waiting` until the token is succeeded or failed. `delivering` while +// the outcome is on its way to the workflow; Volcano finishes the +// delivery on its own. `succeeded` or `failed` once the workflow has the +// outcome. +type DurableCallbackStatus string + +// DurableCallbackToken The `vcb_` token the workflow handed out. It is the only credential +// the callback operations take, so treat it as a secret. +type DurableCallbackToken = string + // DurableExecution defines model for DurableExecution. type DurableExecution struct { // ClientToken Lets a browser read and send the execution's messages. Returned only @@ -7158,6 +7214,16 @@ type ExportProjectSourceRequest struct { ProductionBranch string `json:"production_branch"` } +// FailDurableCallbackRequest defines model for FailDurableCallbackRequest. +type FailDurableCallbackRequest struct { + // Error The error the waiting step raises. + Error DurableCallbackError `json:"error"` + + // Token The `vcb_` token the workflow handed out. It is the only credential + // the callback operations take, so treat it as a secret. + Token DurableCallbackToken `json:"token"` +} + // Frontend defines model for Frontend. type Frontend struct { // AppRoot Optional relative POSIX path from the uploaded archive root to the Next.js app that is deployed. @@ -7823,6 +7889,13 @@ type GitRepositoryReference struct { Id int64 `json:"id"` } +// HeartbeatDurableCallbackRequest defines model for HeartbeatDurableCallbackRequest. +type HeartbeatDurableCallbackRequest struct { + // Token The `vcb_` token the workflow handed out. It is the only credential + // the callback operations take, so treat it as a secret. + Token DurableCallbackToken `json:"token"` +} + // HostedAuthPageType defines model for HostedAuthPageType. type HostedAuthPageType string @@ -10846,6 +10919,18 @@ type StorageVisibilityRequest struct { IsPublic bool `json:"is_public"` } +// SucceedDurableCallbackRequest defines model for SucceedDurableCallbackRequest. +type SucceedDurableCallbackRequest struct { + // Result Any JSON value other than null, up to 64 KiB. The waiting step + // resolves with it. Null is refused because a timeout resolves the + // step with null. + Result json.RawMessage `json:"result"` + + // Token The `vcb_` token the workflow handed out. It is the only credential + // the callback operations take, so treat it as a secret. + Token DurableCallbackToken `json:"token"` +} + // TestEmailRequest When `html_body` or `text_body` is provided the backend renders // those (plus optional `subject`) through html/text templates // against the standard `Data` (ProjectName, Name, SiteURL) @@ -13458,6 +13543,15 @@ type QueryDatabaseUpdateJSONRequestBody = DatabaseUpdateRequest // RequestDurableApprovalJSONRequestBody defines body for RequestDurableApproval for application/json ContentType. type RequestDurableApprovalJSONRequestBody = RequestDurableApprovalRequest +// FailDurableCallbackJSONRequestBody defines body for FailDurableCallback for application/json ContentType. +type FailDurableCallbackJSONRequestBody = FailDurableCallbackRequest + +// HeartbeatDurableCallbackJSONRequestBody defines body for HeartbeatDurableCallback for application/json ContentType. +type HeartbeatDurableCallbackJSONRequestBody = HeartbeatDurableCallbackRequest + +// SucceedDurableCallbackJSONRequestBody defines body for SucceedDurableCallback for application/json ContentType. +type SucceedDurableCallbackJSONRequestBody = SucceedDurableCallbackRequest + // StartDurableExecutionFromApplicationJSONRequestBody defines body for StartDurableExecutionFromApplication for application/json ContentType. type StartDurableExecutionFromApplicationJSONRequestBody = StartDurableExecutionFromApplicationJSONBody @@ -15338,6 +15432,21 @@ type ClientInterface interface { RequestDurableApproval(ctx context.Context, body RequestDurableApprovalJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) + // FailDurableCallbackWithBody request with any body + FailDurableCallbackWithBody(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) + + FailDurableCallback(ctx context.Context, body FailDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) + + // HeartbeatDurableCallbackWithBody request with any body + HeartbeatDurableCallbackWithBody(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) + + HeartbeatDurableCallback(ctx context.Context, body HeartbeatDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) + + // SucceedDurableCallbackWithBody request with any body + SucceedDurableCallbackWithBody(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) + + SucceedDurableCallback(ctx context.Context, body SucceedDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) + // StartDurableExecutionFromApplicationWithBody request with any body StartDurableExecutionFromApplicationWithBody(ctx context.Context, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) @@ -17290,6 +17399,78 @@ func (c *Client) RequestDurableApproval(ctx context.Context, body RequestDurable return c.Client.Do(req) } +func (c *Client) FailDurableCallbackWithBody(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewFailDurableCallbackRequestWithBody(c.Server, contentType, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + +func (c *Client) FailDurableCallback(ctx context.Context, body FailDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewFailDurableCallbackRequest(c.Server, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + +func (c *Client) HeartbeatDurableCallbackWithBody(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewHeartbeatDurableCallbackRequestWithBody(c.Server, contentType, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + +func (c *Client) HeartbeatDurableCallback(ctx context.Context, body HeartbeatDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewHeartbeatDurableCallbackRequest(c.Server, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + +func (c *Client) SucceedDurableCallbackWithBody(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewSucceedDurableCallbackRequestWithBody(c.Server, contentType, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + +func (c *Client) SucceedDurableCallback(ctx context.Context, body SucceedDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*http.Response, error) { + req, err := NewSucceedDurableCallbackRequest(c.Server, body) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + if err := c.applyEditors(ctx, req, reqEditors); err != nil { + return nil, err + } + return c.Client.Do(req) +} + func (c *Client) StartDurableExecutionFromApplicationWithBody(ctx context.Context, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*http.Response, error) { req, err := NewStartDurableExecutionFromApplicationRequestWithBody(c.Server, functionId, params, contentType, body) if err != nil { @@ -23985,34 +24166,27 @@ func NewRequestDurableApprovalRequestWithBody(server string, contentType string, return req, nil } -// NewStartDurableExecutionFromApplicationRequest calls the generic StartDurableExecutionFromApplication builder with application/json body -func NewStartDurableExecutionFromApplicationRequest(server string, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, body StartDurableExecutionFromApplicationJSONRequestBody) (*http.Request, error) { +// NewFailDurableCallbackRequest calls the generic FailDurableCallback builder with application/json body +func NewFailDurableCallbackRequest(server string, body FailDurableCallbackJSONRequestBody) (*http.Request, error) { var bodyReader io.Reader buf, err := json.Marshal(body) if err != nil { return nil, err } bodyReader = bytes.NewReader(buf) - return NewStartDurableExecutionFromApplicationRequestWithBody(server, functionId, params, "application/json", bodyReader) + return NewFailDurableCallbackRequestWithBody(server, "application/json", bodyReader) } -// NewStartDurableExecutionFromApplicationRequestWithBody generates requests for StartDurableExecutionFromApplication with any type of body -func NewStartDurableExecutionFromApplicationRequestWithBody(server string, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, contentType string, body io.Reader) (*http.Request, error) { +// NewFailDurableCallbackRequestWithBody generates requests for FailDurableCallback with any type of body +func NewFailDurableCallbackRequestWithBody(server string, contentType string, body io.Reader) (*http.Request, error) { var err error - var pathParam0 string - - pathParam0, err = runtime.StyleParamWithOptions("simple", false, "functionId", functionId, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationPath, Type: "string", Format: ""}) - if err != nil { - return nil, err - } - serverURL, err := url.Parse(server) if err != nil { return nil, err } - operationPath := fmt.Sprintf("/durable-functions/%s/executions", pathParam0) + operationPath := fmt.Sprintf("/durable-callbacks/fail") if operationPath[0] == '/' { operationPath = "." + operationPath } @@ -24029,48 +24203,70 @@ func NewStartDurableExecutionFromApplicationRequestWithBody(server string, funct req.Header.Add("Content-Type", contentType) - if params != nil { - - if params.XVolcanoExecutionName != nil { - var headerParam0 string - - headerParam0, err = runtime.StyleParamWithOptions("simple", false, "X-Volcano-Execution-Name", *params.XVolcanoExecutionName, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationHeader, Type: "string", Format: ""}) - if err != nil { - return nil, err - } - - req.Header.Set("X-Volcano-Execution-Name", headerParam0) - } + return req, nil +} +// NewHeartbeatDurableCallbackRequest calls the generic HeartbeatDurableCallback builder with application/json body +func NewHeartbeatDurableCallbackRequest(server string, body HeartbeatDurableCallbackJSONRequestBody) (*http.Request, error) { + var bodyReader io.Reader + buf, err := json.Marshal(body) + if err != nil { + return nil, err } - - return req, nil + bodyReader = bytes.NewReader(buf) + return NewHeartbeatDurableCallbackRequestWithBody(server, "application/json", bodyReader) } -// NewListDurableMessagesFromApplicationRequest generates requests for ListDurableMessagesFromApplication -func NewListDurableMessagesFromApplicationRequest(server string, functionId DurableFunctionId, executionId DurableExecutionId, params *ListDurableMessagesFromApplicationParams) (*http.Request, error) { +// NewHeartbeatDurableCallbackRequestWithBody generates requests for HeartbeatDurableCallback with any type of body +func NewHeartbeatDurableCallbackRequestWithBody(server string, contentType string, body io.Reader) (*http.Request, error) { var err error - var pathParam0 string + serverURL, err := url.Parse(server) + if err != nil { + return nil, err + } - pathParam0, err = runtime.StyleParamWithOptions("simple", false, "functionId", functionId, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationPath, Type: "string", Format: ""}) + operationPath := fmt.Sprintf("/durable-callbacks/heartbeat") + if operationPath[0] == '/' { + operationPath = "." + operationPath + } + + queryURL, err := serverURL.Parse(operationPath) if err != nil { return nil, err } - var pathParam1 string + req, err := http.NewRequest(http.MethodPost, queryURL.String(), body) + if err != nil { + return nil, err + } - pathParam1, err = runtime.StyleParamWithOptions("simple", false, "executionId", executionId, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationPath, Type: "string", Format: "uuid"}) + req.Header.Add("Content-Type", contentType) + + return req, nil +} + +// NewSucceedDurableCallbackRequest calls the generic SucceedDurableCallback builder with application/json body +func NewSucceedDurableCallbackRequest(server string, body SucceedDurableCallbackJSONRequestBody) (*http.Request, error) { + var bodyReader io.Reader + buf, err := json.Marshal(body) if err != nil { return nil, err } + bodyReader = bytes.NewReader(buf) + return NewSucceedDurableCallbackRequestWithBody(server, "application/json", bodyReader) +} + +// NewSucceedDurableCallbackRequestWithBody generates requests for SucceedDurableCallback with any type of body +func NewSucceedDurableCallbackRequestWithBody(server string, contentType string, body io.Reader) (*http.Request, error) { + var err error serverURL, err := url.Parse(server) if err != nil { return nil, err } - operationPath := fmt.Sprintf("/durable-functions/%s/executions/%s/messages", pathParam0, pathParam1) + operationPath := fmt.Sprintf("/durable-callbacks/succeed") if operationPath[0] == '/' { operationPath = "." + operationPath } @@ -24080,85 +24276,71 @@ func NewListDurableMessagesFromApplicationRequest(server string, functionId Dura return nil, err } - if params != nil { - // queryValues collects non-styled parameters (passthrough, JSON) - // that are safe to round-trip through url.Values.Encode(). - queryValues := queryURL.Query() - // rawQueryFragments collects pre-encoded query fragments from - // styled parameters, preserving literal commas as delimiters - // per the OpenAPI spec (e.g. "color=blue,black,brown"). - var rawQueryFragments []string - - if params.Direction != nil { - - if queryFrag, err := runtime.StyleParamWithOptions("form", true, "direction", *params.Direction, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "string", Format: ""}); err != nil { - return nil, err - } else { - for _, qp := range strings.Split(queryFrag, "&") { - rawQueryFragments = append(rawQueryFragments, qp) - } - } - - } - - if params.Name != nil { + req, err := http.NewRequest(http.MethodPost, queryURL.String(), body) + if err != nil { + return nil, err + } - if queryFrag, err := runtime.StyleParamWithOptions("form", true, "name", *params.Name, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "string", Format: ""}); err != nil { - return nil, err - } else { - for _, qp := range strings.Split(queryFrag, "&") { - rawQueryFragments = append(rawQueryFragments, qp) - } - } + req.Header.Add("Content-Type", contentType) - } + return req, nil +} - if params.After != nil { +// NewStartDurableExecutionFromApplicationRequest calls the generic StartDurableExecutionFromApplication builder with application/json body +func NewStartDurableExecutionFromApplicationRequest(server string, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, body StartDurableExecutionFromApplicationJSONRequestBody) (*http.Request, error) { + var bodyReader io.Reader + buf, err := json.Marshal(body) + if err != nil { + return nil, err + } + bodyReader = bytes.NewReader(buf) + return NewStartDurableExecutionFromApplicationRequestWithBody(server, functionId, params, "application/json", bodyReader) +} - if queryFrag, err := runtime.StyleParamWithOptions("form", true, "after", *params.After, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "integer", Format: "int64"}); err != nil { - return nil, err - } else { - for _, qp := range strings.Split(queryFrag, "&") { - rawQueryFragments = append(rawQueryFragments, qp) - } - } +// NewStartDurableExecutionFromApplicationRequestWithBody generates requests for StartDurableExecutionFromApplication with any type of body +func NewStartDurableExecutionFromApplicationRequestWithBody(server string, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, contentType string, body io.Reader) (*http.Request, error) { + var err error - } + var pathParam0 string - if params.Limit != nil { + pathParam0, err = runtime.StyleParamWithOptions("simple", false, "functionId", functionId, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationPath, Type: "string", Format: ""}) + if err != nil { + return nil, err + } - if queryFrag, err := runtime.StyleParamWithOptions("form", true, "limit", *params.Limit, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "integer", Format: ""}); err != nil { - return nil, err - } else { - for _, qp := range strings.Split(queryFrag, "&") { - rawQueryFragments = append(rawQueryFragments, qp) - } - } + serverURL, err := url.Parse(server) + if err != nil { + return nil, err + } - } + operationPath := fmt.Sprintf("/durable-functions/%s/executions", pathParam0) + if operationPath[0] == '/' { + operationPath = "." + operationPath + } - if encoded := queryValues.Encode(); encoded != "" { - rawQueryFragments = append(rawQueryFragments, encoded) - } - queryURL.RawQuery = strings.Join(rawQueryFragments, "&") + queryURL, err := serverURL.Parse(operationPath) + if err != nil { + return nil, err } - req, err := http.NewRequest(http.MethodGet, queryURL.String(), nil) + req, err := http.NewRequest(http.MethodPost, queryURL.String(), body) if err != nil { return nil, err } + req.Header.Add("Content-Type", contentType) + if params != nil { - if params.XVolcanoClientToken != nil { + if params.XVolcanoExecutionName != nil { var headerParam0 string - headerParam0, err = runtime.StyleParamWithOptions("simple", false, "X-Volcano-Client-Token", *params.XVolcanoClientToken, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationHeader, Type: "string", Format: ""}) + headerParam0, err = runtime.StyleParamWithOptions("simple", false, "X-Volcano-Execution-Name", *params.XVolcanoExecutionName, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationHeader, Type: "string", Format: ""}) if err != nil { return nil, err } - req.Header.Set("X-Volcano-Client-Token", headerParam0) + req.Header.Set("X-Volcano-Execution-Name", headerParam0) } } @@ -24166,19 +24348,138 @@ func NewListDurableMessagesFromApplicationRequest(server string, functionId Dura return req, nil } -// NewSendDurableMessageFromApplicationRequest calls the generic SendDurableMessageFromApplication builder with application/json body -func NewSendDurableMessageFromApplicationRequest(server string, functionId DurableFunctionId, executionId DurableExecutionId, params *SendDurableMessageFromApplicationParams, body SendDurableMessageFromApplicationJSONRequestBody) (*http.Request, error) { - var bodyReader io.Reader - buf, err := json.Marshal(body) - if err != nil { - return nil, err - } - bodyReader = bytes.NewReader(buf) - return NewSendDurableMessageFromApplicationRequestWithBody(server, functionId, executionId, params, "application/json", bodyReader) -} - -// NewSendDurableMessageFromApplicationRequestWithBody generates requests for SendDurableMessageFromApplication with any type of body -func NewSendDurableMessageFromApplicationRequestWithBody(server string, functionId DurableFunctionId, executionId DurableExecutionId, params *SendDurableMessageFromApplicationParams, contentType string, body io.Reader) (*http.Request, error) { +// NewListDurableMessagesFromApplicationRequest generates requests for ListDurableMessagesFromApplication +func NewListDurableMessagesFromApplicationRequest(server string, functionId DurableFunctionId, executionId DurableExecutionId, params *ListDurableMessagesFromApplicationParams) (*http.Request, error) { + var err error + + var pathParam0 string + + pathParam0, err = runtime.StyleParamWithOptions("simple", false, "functionId", functionId, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationPath, Type: "string", Format: ""}) + if err != nil { + return nil, err + } + + var pathParam1 string + + pathParam1, err = runtime.StyleParamWithOptions("simple", false, "executionId", executionId, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationPath, Type: "string", Format: "uuid"}) + if err != nil { + return nil, err + } + + serverURL, err := url.Parse(server) + if err != nil { + return nil, err + } + + operationPath := fmt.Sprintf("/durable-functions/%s/executions/%s/messages", pathParam0, pathParam1) + if operationPath[0] == '/' { + operationPath = "." + operationPath + } + + queryURL, err := serverURL.Parse(operationPath) + if err != nil { + return nil, err + } + + if params != nil { + // queryValues collects non-styled parameters (passthrough, JSON) + // that are safe to round-trip through url.Values.Encode(). + queryValues := queryURL.Query() + // rawQueryFragments collects pre-encoded query fragments from + // styled parameters, preserving literal commas as delimiters + // per the OpenAPI spec (e.g. "color=blue,black,brown"). + var rawQueryFragments []string + + if params.Direction != nil { + + if queryFrag, err := runtime.StyleParamWithOptions("form", true, "direction", *params.Direction, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "string", Format: ""}); err != nil { + return nil, err + } else { + for _, qp := range strings.Split(queryFrag, "&") { + rawQueryFragments = append(rawQueryFragments, qp) + } + } + + } + + if params.Name != nil { + + if queryFrag, err := runtime.StyleParamWithOptions("form", true, "name", *params.Name, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "string", Format: ""}); err != nil { + return nil, err + } else { + for _, qp := range strings.Split(queryFrag, "&") { + rawQueryFragments = append(rawQueryFragments, qp) + } + } + + } + + if params.After != nil { + + if queryFrag, err := runtime.StyleParamWithOptions("form", true, "after", *params.After, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "integer", Format: "int64"}); err != nil { + return nil, err + } else { + for _, qp := range strings.Split(queryFrag, "&") { + rawQueryFragments = append(rawQueryFragments, qp) + } + } + + } + + if params.Limit != nil { + + if queryFrag, err := runtime.StyleParamWithOptions("form", true, "limit", *params.Limit, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationQuery, Type: "integer", Format: ""}); err != nil { + return nil, err + } else { + for _, qp := range strings.Split(queryFrag, "&") { + rawQueryFragments = append(rawQueryFragments, qp) + } + } + + } + + if encoded := queryValues.Encode(); encoded != "" { + rawQueryFragments = append(rawQueryFragments, encoded) + } + queryURL.RawQuery = strings.Join(rawQueryFragments, "&") + } + + req, err := http.NewRequest(http.MethodGet, queryURL.String(), nil) + if err != nil { + return nil, err + } + + if params != nil { + + if params.XVolcanoClientToken != nil { + var headerParam0 string + + headerParam0, err = runtime.StyleParamWithOptions("simple", false, "X-Volcano-Client-Token", *params.XVolcanoClientToken, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationHeader, Type: "string", Format: ""}) + if err != nil { + return nil, err + } + + req.Header.Set("X-Volcano-Client-Token", headerParam0) + } + + } + + return req, nil +} + +// NewSendDurableMessageFromApplicationRequest calls the generic SendDurableMessageFromApplication builder with application/json body +func NewSendDurableMessageFromApplicationRequest(server string, functionId DurableFunctionId, executionId DurableExecutionId, params *SendDurableMessageFromApplicationParams, body SendDurableMessageFromApplicationJSONRequestBody) (*http.Request, error) { + var bodyReader io.Reader + buf, err := json.Marshal(body) + if err != nil { + return nil, err + } + bodyReader = bytes.NewReader(buf) + return NewSendDurableMessageFromApplicationRequestWithBody(server, functionId, executionId, params, "application/json", bodyReader) +} + +// NewSendDurableMessageFromApplicationRequestWithBody generates requests for SendDurableMessageFromApplication with any type of body +func NewSendDurableMessageFromApplicationRequestWithBody(server string, functionId DurableFunctionId, executionId DurableExecutionId, params *SendDurableMessageFromApplicationParams, contentType string, body io.Reader) (*http.Request, error) { var err error var pathParam0 string @@ -39217,6 +39518,21 @@ type ClientWithResponsesInterface interface { RequestDurableApprovalWithResponse(ctx context.Context, body RequestDurableApprovalJSONRequestBody, reqEditors ...RequestEditorFn) (*RequestDurableApprovalClientResponse, error) + // FailDurableCallbackWithBodyWithResponse request with any body + FailDurableCallbackWithBodyWithResponse(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*FailDurableCallbackClientResponse, error) + + FailDurableCallbackWithResponse(ctx context.Context, body FailDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*FailDurableCallbackClientResponse, error) + + // HeartbeatDurableCallbackWithBodyWithResponse request with any body + HeartbeatDurableCallbackWithBodyWithResponse(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*HeartbeatDurableCallbackClientResponse, error) + + HeartbeatDurableCallbackWithResponse(ctx context.Context, body HeartbeatDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*HeartbeatDurableCallbackClientResponse, error) + + // SucceedDurableCallbackWithBodyWithResponse request with any body + SucceedDurableCallbackWithBodyWithResponse(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*SucceedDurableCallbackClientResponse, error) + + SucceedDurableCallbackWithResponse(ctx context.Context, body SucceedDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*SucceedDurableCallbackClientResponse, error) + // StartDurableExecutionFromApplicationWithBodyWithResponse request with any body StartDurableExecutionFromApplicationWithBodyWithResponse(ctx context.Context, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*StartDurableExecutionFromApplicationClientResponse, error) @@ -41965,6 +42281,117 @@ func (r RequestDurableApprovalClientResponse) ContentType() string { return "" } +type FailDurableCallbackClientResponse struct { + Body []byte + HTTPResponse *http.Response + JSON200 *DurableCallbackOutcome + JSON202 *DurableCallbackOutcome + JSON400 *Error + JSON404 *Error + JSON409 *Error + JSON410 *Error + JSON413 *Error + JSON429 *Error +} + +// Status returns HTTPResponse.Status +func (r FailDurableCallbackClientResponse) Status() string { + if r.HTTPResponse != nil { + return r.HTTPResponse.Status + } + return http.StatusText(0) +} + +// StatusCode returns HTTPResponse.StatusCode +func (r FailDurableCallbackClientResponse) StatusCode() int { + if r.HTTPResponse != nil { + return r.HTTPResponse.StatusCode + } + return 0 +} + +// ContentType is a convenience method to retrieve the Content-Type value from the HTTP response headers +func (r FailDurableCallbackClientResponse) ContentType() string { + if r.HTTPResponse != nil { + return r.HTTPResponse.Header.Get("Content-Type") + } + return "" +} + +type HeartbeatDurableCallbackClientResponse struct { + Body []byte + HTTPResponse *http.Response + JSON200 *DurableCallbackOutcome + JSON400 *Error + JSON404 *Error + JSON409 *Error + JSON410 *Error + JSON413 *Error + JSON429 *Error + JSON503 *Error +} + +// Status returns HTTPResponse.Status +func (r HeartbeatDurableCallbackClientResponse) Status() string { + if r.HTTPResponse != nil { + return r.HTTPResponse.Status + } + return http.StatusText(0) +} + +// StatusCode returns HTTPResponse.StatusCode +func (r HeartbeatDurableCallbackClientResponse) StatusCode() int { + if r.HTTPResponse != nil { + return r.HTTPResponse.StatusCode + } + return 0 +} + +// ContentType is a convenience method to retrieve the Content-Type value from the HTTP response headers +func (r HeartbeatDurableCallbackClientResponse) ContentType() string { + if r.HTTPResponse != nil { + return r.HTTPResponse.Header.Get("Content-Type") + } + return "" +} + +type SucceedDurableCallbackClientResponse struct { + Body []byte + HTTPResponse *http.Response + JSON200 *DurableCallbackOutcome + JSON202 *DurableCallbackOutcome + JSON400 *Error + JSON404 *Error + JSON409 *Error + JSON410 *Error + JSON413 *Error + JSON429 *Error +} + +// Status returns HTTPResponse.Status +func (r SucceedDurableCallbackClientResponse) Status() string { + if r.HTTPResponse != nil { + return r.HTTPResponse.Status + } + return http.StatusText(0) +} + +// StatusCode returns HTTPResponse.StatusCode +func (r SucceedDurableCallbackClientResponse) StatusCode() int { + if r.HTTPResponse != nil { + return r.HTTPResponse.StatusCode + } + return 0 +} + +// ContentType is a convenience method to retrieve the Content-Type value from the HTTP response headers +func (r SucceedDurableCallbackClientResponse) ContentType() string { + if r.HTTPResponse != nil { + return r.HTTPResponse.Header.Get("Content-Type") + } + return "" +} + type StartDurableExecutionFromApplicationClientResponse struct { Body []byte HTTPResponse *http.Response @@ -51730,6 +52157,57 @@ func (c *ClientWithResponses) RequestDurableApprovalWithResponse(ctx context.Con return ParseRequestDurableApprovalClientResponse(rsp) } +// FailDurableCallbackWithBodyWithResponse request with arbitrary body returning *FailDurableCallbackClientResponse +func (c *ClientWithResponses) FailDurableCallbackWithBodyWithResponse(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*FailDurableCallbackClientResponse, error) { + rsp, err := c.FailDurableCallbackWithBody(ctx, contentType, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseFailDurableCallbackClientResponse(rsp) +} + +func (c *ClientWithResponses) FailDurableCallbackWithResponse(ctx context.Context, body FailDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*FailDurableCallbackClientResponse, error) { + rsp, err := c.FailDurableCallback(ctx, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseFailDurableCallbackClientResponse(rsp) +} + +// HeartbeatDurableCallbackWithBodyWithResponse request with arbitrary body returning *HeartbeatDurableCallbackClientResponse +func (c *ClientWithResponses) HeartbeatDurableCallbackWithBodyWithResponse(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*HeartbeatDurableCallbackClientResponse, error) { + rsp, err := c.HeartbeatDurableCallbackWithBody(ctx, contentType, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseHeartbeatDurableCallbackClientResponse(rsp) +} + +func (c *ClientWithResponses) HeartbeatDurableCallbackWithResponse(ctx context.Context, body HeartbeatDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*HeartbeatDurableCallbackClientResponse, error) { + rsp, err := c.HeartbeatDurableCallback(ctx, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseHeartbeatDurableCallbackClientResponse(rsp) +} + +// SucceedDurableCallbackWithBodyWithResponse request with arbitrary body returning *SucceedDurableCallbackClientResponse +func (c *ClientWithResponses) SucceedDurableCallbackWithBodyWithResponse(ctx context.Context, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*SucceedDurableCallbackClientResponse, error) { + rsp, err := c.SucceedDurableCallbackWithBody(ctx, contentType, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseSucceedDurableCallbackClientResponse(rsp) +} + +func (c *ClientWithResponses) SucceedDurableCallbackWithResponse(ctx context.Context, body SucceedDurableCallbackJSONRequestBody, reqEditors ...RequestEditorFn) (*SucceedDurableCallbackClientResponse, error) { + rsp, err := c.SucceedDurableCallback(ctx, body, reqEditors...) + if err != nil { + return nil, err + } + return ParseSucceedDurableCallbackClientResponse(rsp) +} + // StartDurableExecutionFromApplicationWithBodyWithResponse request with arbitrary body returning *StartDurableExecutionFromApplicationClientResponse func (c *ClientWithResponses) StartDurableExecutionFromApplicationWithBodyWithResponse(ctx context.Context, functionId DurableFunctionId, params *StartDurableExecutionFromApplicationParams, contentType string, body io.Reader, reqEditors ...RequestEditorFn) (*StartDurableExecutionFromApplicationClientResponse, error) { rsp, err := c.StartDurableExecutionFromApplicationWithBody(ctx, functionId, params, contentType, body, reqEditors...) @@ -56473,27 +56951,353 @@ func ParseQueryDatabaseBranchDeleteClientResponse(rsp *http.Response) (*QueryDat } response.JSON429 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: - var dest DatabaseBranchQueryUnavailable - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON503 = &dest - + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: + var dest DatabaseBranchQueryUnavailable + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON503 = &dest + + } + + return response, nil +} + +// ParseQueryDatabaseBranchInsertClientResponse parses an HTTP response from a QueryDatabaseBranchInsertWithResponse call +func ParseQueryDatabaseBranchInsertClientResponse(rsp *http.Response) (*QueryDatabaseBranchInsertClientResponse, error) { + bodyBytes, err := io.ReadAll(rsp.Body) + defer func() { _ = rsp.Body.Close() }() + if err != nil { + return nil, err + } + + response := &QueryDatabaseBranchInsertClientResponse{ + Body: bodyBytes, + HTTPResponse: rsp, + } + + switch { + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: + var dest DatabaseQueryResult + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON200 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON400 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON401 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON403 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON404 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: + var dest DatabaseQueryCapExceeded + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON429 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: + var dest DatabaseBranchQueryUnavailable + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON503 = &dest + + } + + return response, nil +} + +// ParseQueryDatabaseBranchPingClientResponse parses an HTTP response from a QueryDatabaseBranchPingWithResponse call +func ParseQueryDatabaseBranchPingClientResponse(rsp *http.Response) (*QueryDatabaseBranchPingClientResponse, error) { + bodyBytes, err := io.ReadAll(rsp.Body) + defer func() { _ = rsp.Body.Close() }() + if err != nil { + return nil, err + } + + response := &QueryDatabaseBranchPingClientResponse{ + Body: bodyBytes, + HTTPResponse: rsp, + } + + switch { + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: + var dest DatabaseQueryResult + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON200 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON401 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON403 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON404 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: + var dest DatabaseQueryCapExceeded + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON429 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: + var dest DatabaseBranchQueryUnavailable + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON503 = &dest + + } + + return response, nil +} + +// ParseQueryDatabaseBranchSelectClientResponse parses an HTTP response from a QueryDatabaseBranchSelectWithResponse call +func ParseQueryDatabaseBranchSelectClientResponse(rsp *http.Response) (*QueryDatabaseBranchSelectClientResponse, error) { + bodyBytes, err := io.ReadAll(rsp.Body) + defer func() { _ = rsp.Body.Close() }() + if err != nil { + return nil, err + } + + response := &QueryDatabaseBranchSelectClientResponse{ + Body: bodyBytes, + HTTPResponse: rsp, + } + + switch { + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: + var dest DatabaseQueryResult + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON200 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON400 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON401 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON403 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON404 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: + var dest DatabaseQueryCapExceeded + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON429 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: + var dest DatabaseBranchQueryUnavailable + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON503 = &dest + + } + + return response, nil +} + +// ParseQueryDatabaseBranchUpdateClientResponse parses an HTTP response from a QueryDatabaseBranchUpdateWithResponse call +func ParseQueryDatabaseBranchUpdateClientResponse(rsp *http.Response) (*QueryDatabaseBranchUpdateClientResponse, error) { + bodyBytes, err := io.ReadAll(rsp.Body) + defer func() { _ = rsp.Body.Close() }() + if err != nil { + return nil, err + } + + response := &QueryDatabaseBranchUpdateClientResponse{ + Body: bodyBytes, + HTTPResponse: rsp, + } + + switch { + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: + var dest DatabaseQueryResult + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON200 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON400 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON401 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON403 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON404 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: + var dest DatabaseQueryCapExceeded + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON429 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: + var dest DatabaseBranchQueryUnavailable + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON503 = &dest + + } + + return response, nil +} + +// ParseQueryDatabaseDeleteClientResponse parses an HTTP response from a QueryDatabaseDeleteWithResponse call +func ParseQueryDatabaseDeleteClientResponse(rsp *http.Response) (*QueryDatabaseDeleteClientResponse, error) { + bodyBytes, err := io.ReadAll(rsp.Body) + defer func() { _ = rsp.Body.Close() }() + if err != nil { + return nil, err + } + + response := &QueryDatabaseDeleteClientResponse{ + Body: bodyBytes, + HTTPResponse: rsp, + } + + switch { + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: + var dest DatabaseQueryResult + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON200 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON400 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON401 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON403 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON404 = &dest + + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: + var dest DatabaseQueryCapExceeded + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON429 = &dest + } return response, nil } -// ParseQueryDatabaseBranchInsertClientResponse parses an HTTP response from a QueryDatabaseBranchInsertWithResponse call -func ParseQueryDatabaseBranchInsertClientResponse(rsp *http.Response) (*QueryDatabaseBranchInsertClientResponse, error) { +// ParseQueryDatabaseInsertClientResponse parses an HTTP response from a QueryDatabaseInsertWithResponse call +func ParseQueryDatabaseInsertClientResponse(rsp *http.Response) (*QueryDatabaseInsertClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabaseBranchInsertClientResponse{ + response := &QueryDatabaseInsertClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } @@ -56541,27 +57345,20 @@ func ParseQueryDatabaseBranchInsertClientResponse(rsp *http.Response) (*QueryDat } response.JSON429 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: - var dest DatabaseBranchQueryUnavailable - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON503 = &dest - } return response, nil } -// ParseQueryDatabaseBranchPingClientResponse parses an HTTP response from a QueryDatabaseBranchPingWithResponse call -func ParseQueryDatabaseBranchPingClientResponse(rsp *http.Response) (*QueryDatabaseBranchPingClientResponse, error) { +// ParseQueryDatabasePingClientResponse parses an HTTP response from a QueryDatabasePingWithResponse call +func ParseQueryDatabasePingClientResponse(rsp *http.Response) (*QueryDatabasePingClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabaseBranchPingClientResponse{ + response := &QueryDatabasePingClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } @@ -56602,27 +57399,20 @@ func ParseQueryDatabaseBranchPingClientResponse(rsp *http.Response) (*QueryDatab } response.JSON429 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: - var dest DatabaseBranchQueryUnavailable - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON503 = &dest - } return response, nil } -// ParseQueryDatabaseBranchSelectClientResponse parses an HTTP response from a QueryDatabaseBranchSelectWithResponse call -func ParseQueryDatabaseBranchSelectClientResponse(rsp *http.Response) (*QueryDatabaseBranchSelectClientResponse, error) { +// ParseQueryDatabaseSelectClientResponse parses an HTTP response from a QueryDatabaseSelectWithResponse call +func ParseQueryDatabaseSelectClientResponse(rsp *http.Response) (*QueryDatabaseSelectClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabaseBranchSelectClientResponse{ + response := &QueryDatabaseSelectClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } @@ -56670,27 +57460,20 @@ func ParseQueryDatabaseBranchSelectClientResponse(rsp *http.Response) (*QueryDat } response.JSON429 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: - var dest DatabaseBranchQueryUnavailable - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON503 = &dest - } return response, nil } -// ParseQueryDatabaseBranchUpdateClientResponse parses an HTTP response from a QueryDatabaseBranchUpdateWithResponse call -func ParseQueryDatabaseBranchUpdateClientResponse(rsp *http.Response) (*QueryDatabaseBranchUpdateClientResponse, error) { +// ParseQueryDatabaseUpdateClientResponse parses an HTTP response from a QueryDatabaseUpdateWithResponse call +func ParseQueryDatabaseUpdateClientResponse(rsp *http.Response) (*QueryDatabaseUpdateClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabaseBranchUpdateClientResponse{ + response := &QueryDatabaseUpdateClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } @@ -56738,34 +57521,27 @@ func ParseQueryDatabaseBranchUpdateClientResponse(rsp *http.Response) (*QueryDat } response.JSON429 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: - var dest DatabaseBranchQueryUnavailable - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON503 = &dest - } return response, nil } -// ParseQueryDatabaseDeleteClientResponse parses an HTTP response from a QueryDatabaseDeleteWithResponse call -func ParseQueryDatabaseDeleteClientResponse(rsp *http.Response) (*QueryDatabaseDeleteClientResponse, error) { +// ParseListDeploymentsClientResponse parses an HTTP response from a ListDeploymentsWithResponse call +func ParseListDeploymentsClientResponse(rsp *http.Response) (*ListDeploymentsClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabaseDeleteClientResponse{ + response := &ListDeploymentsClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } switch { case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: - var dest DatabaseQueryResult + var dest PaginatedProjectDeployments if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } @@ -56792,46 +57568,46 @@ func ParseQueryDatabaseDeleteClientResponse(rsp *http.Response) (*QueryDatabaseD } response.JSON403 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 500: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON404 = &dest - - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: - var dest DatabaseQueryCapExceeded - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON429 = &dest + response.JSON500 = &dest } return response, nil } -// ParseQueryDatabaseInsertClientResponse parses an HTTP response from a QueryDatabaseInsertWithResponse call -func ParseQueryDatabaseInsertClientResponse(rsp *http.Response) (*QueryDatabaseInsertClientResponse, error) { +// ParseRequestDurableApprovalClientResponse parses an HTTP response from a RequestDurableApprovalWithResponse call +func ParseRequestDurableApprovalClientResponse(rsp *http.Response) (*RequestDurableApprovalClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabaseInsertClientResponse{ + response := &RequestDurableApprovalClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } switch { case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: - var dest DatabaseQueryResult + var dest DurableApprovalRegistration if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } response.JSON200 = &dest + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 201: + var dest DurableApprovalRegistration + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON201 = &dest + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { @@ -56839,73 +57615,80 @@ func ParseQueryDatabaseInsertClientResponse(rsp *http.Response) (*QueryDatabaseI } response.JSON400 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON401 = &dest + response.JSON404 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 409: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON403 = &dest + response.JSON409 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 413: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON404 = &dest + response.JSON413 = &dest case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: - var dest DatabaseQueryCapExceeded + var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } response.JSON429 = &dest + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: + var dest Error + if err := json.Unmarshal(bodyBytes, &dest); err != nil { + return nil, err + } + response.JSON503 = &dest + } return response, nil } -// ParseQueryDatabasePingClientResponse parses an HTTP response from a QueryDatabasePingWithResponse call -func ParseQueryDatabasePingClientResponse(rsp *http.Response) (*QueryDatabasePingClientResponse, error) { +// ParseFailDurableCallbackClientResponse parses an HTTP response from a FailDurableCallbackWithResponse call +func ParseFailDurableCallbackClientResponse(rsp *http.Response) (*FailDurableCallbackClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabasePingClientResponse{ + response := &FailDurableCallbackClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } switch { case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: - var dest DatabaseQueryResult + var dest DurableCallbackOutcome if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } response.JSON200 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: - var dest Error + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 202: + var dest DurableCallbackOutcome if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON401 = &dest + response.JSON202 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON403 = &dest + response.JSON400 = &dest case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: var dest Error @@ -56914,69 +57697,29 @@ func ParseQueryDatabasePingClientResponse(rsp *http.Response) (*QueryDatabasePin } response.JSON404 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: - var dest DatabaseQueryCapExceeded - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON429 = &dest - - } - - return response, nil -} - -// ParseQueryDatabaseSelectClientResponse parses an HTTP response from a QueryDatabaseSelectWithResponse call -func ParseQueryDatabaseSelectClientResponse(rsp *http.Response) (*QueryDatabaseSelectClientResponse, error) { - bodyBytes, err := io.ReadAll(rsp.Body) - defer func() { _ = rsp.Body.Close() }() - if err != nil { - return nil, err - } - - response := &QueryDatabaseSelectClientResponse{ - Body: bodyBytes, - HTTPResponse: rsp, - } - - switch { - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: - var dest DatabaseQueryResult - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON200 = &dest - - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: - var dest Error - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON400 = &dest - - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 409: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON401 = &dest + response.JSON409 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 410: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON403 = &dest + response.JSON410 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 413: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON404 = &dest + response.JSON413 = &dest case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: - var dest DatabaseQueryCapExceeded + var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } @@ -56987,22 +57730,22 @@ func ParseQueryDatabaseSelectClientResponse(rsp *http.Response) (*QueryDatabaseS return response, nil } -// ParseQueryDatabaseUpdateClientResponse parses an HTTP response from a QueryDatabaseUpdateWithResponse call -func ParseQueryDatabaseUpdateClientResponse(rsp *http.Response) (*QueryDatabaseUpdateClientResponse, error) { +// ParseHeartbeatDurableCallbackClientResponse parses an HTTP response from a HeartbeatDurableCallbackWithResponse call +func ParseHeartbeatDurableCallbackClientResponse(rsp *http.Response) (*HeartbeatDurableCallbackClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &QueryDatabaseUpdateClientResponse{ + response := &HeartbeatDurableCallbackClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } switch { case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: - var dest DatabaseQueryResult + var dest DurableCallbackOutcome if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } @@ -57015,20 +57758,6 @@ func ParseQueryDatabaseUpdateClientResponse(rsp *http.Response) (*QueryDatabaseU } response.JSON400 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: - var dest Error - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON401 = &dest - - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: - var dest Error - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON403 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 404: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { @@ -57036,99 +57765,73 @@ func ParseQueryDatabaseUpdateClientResponse(rsp *http.Response) (*QueryDatabaseU } response.JSON404 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: - var dest DatabaseQueryCapExceeded - if err := json.Unmarshal(bodyBytes, &dest); err != nil { - return nil, err - } - response.JSON429 = &dest - - } - - return response, nil -} - -// ParseListDeploymentsClientResponse parses an HTTP response from a ListDeploymentsWithResponse call -func ParseListDeploymentsClientResponse(rsp *http.Response) (*ListDeploymentsClientResponse, error) { - bodyBytes, err := io.ReadAll(rsp.Body) - defer func() { _ = rsp.Body.Close() }() - if err != nil { - return nil, err - } - - response := &ListDeploymentsClientResponse{ - Body: bodyBytes, - HTTPResponse: rsp, - } - - switch { - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: - var dest PaginatedProjectDeployments + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 409: + var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON200 = &dest + response.JSON409 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 410: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON400 = &dest + response.JSON410 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 401: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 413: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON401 = &dest + response.JSON413 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 403: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON403 = &dest + response.JSON429 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 500: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON500 = &dest + response.JSON503 = &dest } return response, nil } -// ParseRequestDurableApprovalClientResponse parses an HTTP response from a RequestDurableApprovalWithResponse call -func ParseRequestDurableApprovalClientResponse(rsp *http.Response) (*RequestDurableApprovalClientResponse, error) { +// ParseSucceedDurableCallbackClientResponse parses an HTTP response from a SucceedDurableCallbackWithResponse call +func ParseSucceedDurableCallbackClientResponse(rsp *http.Response) (*SucceedDurableCallbackClientResponse, error) { bodyBytes, err := io.ReadAll(rsp.Body) defer func() { _ = rsp.Body.Close() }() if err != nil { return nil, err } - response := &RequestDurableApprovalClientResponse{ + response := &SucceedDurableCallbackClientResponse{ Body: bodyBytes, HTTPResponse: rsp, } switch { case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 200: - var dest DurableApprovalRegistration + var dest DurableCallbackOutcome if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } response.JSON200 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 201: - var dest DurableApprovalRegistration + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 202: + var dest DurableCallbackOutcome if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON201 = &dest + response.JSON202 = &dest case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 400: var dest Error @@ -57151,26 +57854,26 @@ func ParseRequestDurableApprovalClientResponse(rsp *http.Response) (*RequestDura } response.JSON409 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 413: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 410: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON413 = &dest + response.JSON410 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 413: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON429 = &dest + response.JSON413 = &dest - case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 503: + case strings.Contains(rsp.Header.Get("Content-Type"), "json") && rsp.StatusCode == 429: var dest Error if err := json.Unmarshal(bodyBytes, &dest); err != nil { return nil, err } - response.JSON503 = &dest + response.JSON429 = &dest } diff --git a/internal/cmd/durable/callbacks/callbacks.go b/internal/cmd/durable/callbacks/callbacks.go new file mode 100644 index 00000000..6632967b --- /dev/null +++ b/internal/cmd/durable/callbacks/callbacks.go @@ -0,0 +1,83 @@ +// Package callbacks wires the volcano durable callbacks subcommands. +package callbacks + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "strings" + + "github.com/spf13/cobra" + + "github.com/Kong/volcano-cli/internal/apiclient" + "github.com/Kong/volcano-cli/internal/config" + "github.com/Kong/volcano-cli/internal/output" + cliruntime "github.com/Kong/volcano-cli/internal/runtime" +) + +// maxTokenInput bounds what is read from stdin for a token, which is far +// shorter. +const maxTokenInput = 4096 + +// New returns the durable callbacks command. +func New(deps cliruntime.Deps) *cobra.Command { + cmd := &cobra.Command{ + Use: "callbacks", + Short: "Answer durable callbacks with their token", + Long: `A callback is a wait a durable workflow hands to another system. The workflow +calls ctx.waitForCallback, passes the token it gets to that system, and suspends +until the token is succeeded or failed, or its timeout passes. + +These commands are that system's side. The token is the only credential they +send, so they need no login or project: whoever holds the token can answer the +wait. Treat it as a secret. Pass - in place of the token to read it from stdin, +which keeps it out of your shell history and the process list. + +A token takes one answer. Once it has succeeded or failed, every later answer +or heartbeat is refused, including a retry of the answer that went through.`, + } + // Answering needs no project, so a project link that can't be used must not + // stop it. + if deps.ConfigLoader == nil { + deps.ConfigLoader = config.LoadWithoutLink + } + cmd.AddCommand(newSucceed(deps)) + cmd.AddCommand(newFail(deps)) + cmd.AddCommand(newHeartbeat(deps)) + return cmd +} + +// readToken returns the token argument, or the token on stdin when the +// argument is "-". +func readToken(arg string, in io.Reader) (string, error) { + if arg != "-" { + return strings.TrimSpace(arg), nil + } + data, err := io.ReadAll(io.LimitReader(in, maxTokenInput)) + if err != nil { + return "", fmt.Errorf("failed to read the callback token from stdin: %w", err) + } + token := strings.TrimSpace(string(data)) + if token == "" { + return "", errors.New("no callback token on stdin") + } + return token, nil +} + +// reportAnswer prints what became of a succeed or fail. delivering is not a +// failure: the outcome is recorded, and sending it again would be refused. +func reportAnswer(w io.Writer, outcome *apiclient.DurableCallbackOutcome, done string) { + if outcome.Status == apiclient.DurableCallbackStatusDelivering { + output.Success(w, "Volcano recorded the answer and is still delivering it to the execution; do not send it again") + return + } + output.Success(w, "%s", done) +} + +// writeJSON prints value as indented JSON for --json consumers. +func writeJSON(w io.Writer, value any) error { + encoder := json.NewEncoder(w) + encoder.SetIndent("", " ") + return encoder.Encode(value) +} diff --git a/internal/cmd/durable/callbacks/callbacks_test.go b/internal/cmd/durable/callbacks/callbacks_test.go new file mode 100644 index 00000000..3ae79835 --- /dev/null +++ b/internal/cmd/durable/callbacks/callbacks_test.go @@ -0,0 +1,347 @@ +package callbacks + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "github.com/spf13/cobra" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Kong/volcano-cli/internal/api" + cliconfig "github.com/Kong/volcano-cli/internal/config" + cliruntime "github.com/Kong/volcano-cli/internal/runtime" +) + +const ( + projectID = "22222222-2222-4222-8222-222222222222" + token = "vcb_0000000000000000000000000000000000000000000" +) + +// The CLI is logged in, but the answer carries only the token: the routes +// take no other credential, and the holder may not be a project member. +func TestSucceedSendsTheTokenAndResultAndNoCredential(t *testing.T) { + setCallbacksTestHome(t) + var body []byte + var header http.Header + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodPost, r.Method) + assert.Equal(t, "/durable-callbacks/succeed", r.URL.Path) + header = r.Header.Clone() + var err error + body, err = io.ReadAll(r.Body) + assert.NoError(t, err) + respondJSON(t, w, http.StatusOK, map[string]any{"status": "succeeded"}) + })) + defer server.Close() + + out, err := executeCommand(t, newCallbacksCommand(server), "", + "succeed", token, "--result", `{"amount":12345678901234567890}`) + require.NoError(t, err) + + assert.JSONEq(t, `{"token":"`+token+`","result":{"amount":12345678901234567890}}`, string(body), + "the result is sent as written, large numbers included") + assert.Empty(t, header.Get("Authorization")) + assert.Contains(t, out, "Succeeded the callback; the execution resumes with the result") + assert.NotContains(t, out, token, "the token is not printed back") +} + +// The system a workflow hands a token to has never logged in. +func TestAnswersNeedNoLoginOrProject(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("VOLCANO_TOKEN", "") + t.Setenv("VOLCANO_PROJECT_ID", "") + t.Setenv("VOLCANO_API_URL", "") + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "/durable-callbacks/heartbeat", r.URL.Path) + respondJSON(t, w, http.StatusOK, map[string]any{"status": "waiting"}) + })) + defer server.Close() + + out, err := executeCommand(t, newCallbacksCommand(server), "", "heartbeat", token) + require.NoError(t, err) + assert.Contains(t, out, "Heartbeat sent; the callback is still waiting") +} + +func TestAnswersIgnoreABrokenProjectLink(t *testing.T) { + setCallbacksTestHome(t) + dir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(dir, ".volcano"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, ".volcano", "project.json"), []byte("{not-json"), 0o644)) + t.Chdir(dir) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + respondJSON(t, w, http.StatusOK, map[string]any{"status": "waiting"}) + })) + defer server.Close() + + out, err := executeCommand(t, newCallbacksCommand(server), "", "heartbeat", token) + require.NoError(t, err) + assert.Contains(t, out, "Heartbeat sent; the callback is still waiting") +} + +func TestSucceedReadsTheResultFromEitherFlag(t *testing.T) { + resultFile := filepath.Join(t.TempDir(), "charge.json") + require.NoError(t, os.WriteFile(resultFile, []byte("[\"ch_1\"]\n"), 0o600)) + + for _, tc := range []struct { + name string + args []string + want string + }{ + {name: "inline string", args: []string{"--result", `"done"`}, want: `"done"`}, + {name: "inline false", args: []string{"--result", "false"}, want: `false`}, + {name: "file", args: []string{"--result-file", resultFile}, want: `["ch_1"]`}, + } { + t.Run(tc.name, func(t *testing.T) { + setCallbacksTestHome(t) + var body []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var err error + body, err = io.ReadAll(r.Body) + assert.NoError(t, err) + respondJSON(t, w, http.StatusOK, map[string]any{"status": "succeeded"}) + })) + defer server.Close() + + _, err := executeCommand(t, newCallbacksCommand(server), "", append([]string{"succeed", token}, tc.args...)...) + require.NoError(t, err) + assert.JSONEq(t, `{"token":"`+token+`","result":`+tc.want+`}`, string(body)) + }) + } +} + +func TestFailSendsTheError(t *testing.T) { + for _, tc := range []struct { + name string + args []string + want string + }{ + { + name: "with a message", args: []string{"--type", "CardDeclined", "--message", "Insufficient funds"}, + want: `{"type":"CardDeclined","message":"Insufficient funds"}`, + }, + {name: "type only", args: []string{"--type", "CardDeclined"}, want: `{"type":"CardDeclined"}`}, + } { + t.Run(tc.name, func(t *testing.T) { + setCallbacksTestHome(t) + var body []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "/durable-callbacks/fail", r.URL.Path) + var err error + body, err = io.ReadAll(r.Body) + assert.NoError(t, err) + respondJSON(t, w, http.StatusOK, map[string]any{"status": "failed"}) + })) + defer server.Close() + + out, err := executeCommand(t, newCallbacksCommand(server), "", append([]string{"fail", token}, tc.args...)...) + require.NoError(t, err) + assert.JSONEq(t, `{"token":"`+token+`","error":`+tc.want+`}`, string(body)) + assert.Contains(t, out, "Failed the callback; the execution resumes with the error") + }) + } +} + +// A token piped in stays out of shell history and the process list. +func TestEachCommandReadsTheTokenFromStdin(t *testing.T) { + for _, args := range [][]string{ + {"succeed", "-", "--result", "1"}, + {"fail", "-", "--type", "CardDeclined"}, + {"heartbeat", "-"}, + } { + t.Run(args[0], func(t *testing.T) { + setCallbacksTestHome(t) + var sent struct { + Token string `json:"token"` + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.NoError(t, json.NewDecoder(r.Body).Decode(&sent)) + respondJSON(t, w, http.StatusOK, map[string]any{"status": "waiting"}) + })) + defer server.Close() + + _, err := executeCommand(t, newCallbacksCommand(server), " "+token+"\n", args...) + require.NoError(t, err) + assert.Equal(t, token, sent.Token) + }) + } +} + +// Volcano recorded the answer but has not handed it over yet. That is not a +// failure, and sending it again would be refused. +func TestAnAnswerStillBeingDeliveredSaysNotToResend(t *testing.T) { + for _, args := range [][]string{ + {"succeed", token, "--result", "1"}, + {"fail", token, "--type", "CardDeclined"}, + } { + t.Run(args[0], func(t *testing.T) { + setCallbacksTestHome(t) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + respondJSON(t, w, http.StatusAccepted, map[string]any{"status": "delivering"}) + })) + defer server.Close() + + out, err := executeCommand(t, newCallbacksCommand(server), "", args...) + require.NoError(t, err) + assert.Contains(t, out, "Volcano recorded the answer and is still delivering it to the execution; do not send it again") + assert.NotContains(t, out, "the execution resumes") + }) + } +} + +// What a script reads from stdout under --json is the outcome alone. +func TestJSONPrintsOnlyTheOutcome(t *testing.T) { + setCallbacksTestHome(t) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + respondJSON(t, w, http.StatusAccepted, map[string]any{"status": "delivering"}) + })) + defer server.Close() + + cmd := newCallbacksCommand(server) + var stdout bytes.Buffer + cmd.SetOut(&stdout) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"succeed", token, "--result", "1", "--json"}) + require.NoError(t, cmd.Execute()) + assert.JSONEq(t, `{"status":"delivering"}`, stdout.String()) +} + +// Each of these is refused before any request, with a message that names the +// mistake and never quotes a token. +func TestRefusesBadArgumentsBeforeTheRequest(t *testing.T) { + const almostAToken = "vcb_0000000000000000000000/00000000000000000" + missingFile := filepath.Join(t.TempDir(), "missing.json") + for _, tc := range []struct { + name string + stdin string + args []string + want string + }{ + {name: "malformed token", args: []string{"heartbeat", almostAToken}, want: "invalid callback token"}, + {name: "token without its prefix", args: []string{"heartbeat", "0000000000"}, want: "invalid callback token"}, + {name: "empty stdin", args: []string{"heartbeat", "-"}, want: "no callback token on stdin"}, + {name: "null result", args: []string{"succeed", token, "--result", " null "}, want: "result cannot be null"}, + {name: "malformed result", args: []string{"succeed", token, "--result", "{"}, want: "result must be JSON"}, + {name: "no result", args: []string{"succeed", token}, want: "at least one of the flags in the group [result result-file] is required"}, + {name: "both result flags", args: []string{"succeed", token, "--result", "1", "--result-file", missingFile}, want: "none of the others can be"}, + {name: "missing result file", args: []string{"succeed", token, "--result-file", missingFile}, want: "failed to read result file"}, + {name: "no type", args: []string{"fail", token}, want: `required flag(s) "type" not set`}, + {name: "blank type", args: []string{"fail", token, "--type", " "}, want: "--type needs a name the workflow can branch on"}, + } { + t.Run(tc.name, func(t *testing.T) { + setCallbacksTestHome(t) + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + t.Error("no request may be sent") + })) + defer server.Close() + + _, err := executeCommand(t, newCallbacksCommand(server), tc.stdin, tc.args...) + require.Error(t, err) + assert.Contains(t, err.Error(), tc.want) + assert.NotContains(t, err.Error(), almostAToken) + }) + } +} + +// A holder decides from the refusal whether to retry, so each says what became +// of the wait. +func TestExplainsARefusal(t *testing.T) { + for _, tc := range []struct { + name string + args []string + status int + body map[string]any + want string + }{ + { + name: "unknown token", args: []string{"heartbeat", token}, status: http.StatusNotFound, + body: map[string]any{"error": "callback not found", "code": "callback_not_found"}, + want: "no callback has this token", + }, + { + name: "answered", args: []string{"succeed", token, "--result", "1"}, status: http.StatusConflict, + body: map[string]any{"error": "callback already has an outcome", "code": "callback_completed"}, + want: "the callback already has an outcome; a token takes one answer", + }, + { + name: "timed out", args: []string{"fail", token, "--type", "CardDeclined"}, status: http.StatusGone, + body: map[string]any{"error": "callback expired", "code": "callback_expired"}, + want: "the callback timed out; nothing will take an answer, so do not retry", + }, + { + name: "execution ended", args: []string{"heartbeat", token}, status: http.StatusGone, + body: map[string]any{"error": "durable execution ended", "code": "execution_ended"}, + want: "the callback's execution has ended; nothing will take an answer, so do not retry", + }, + { + name: "heartbeat unavailable", args: []string{"heartbeat", token}, status: http.StatusServiceUnavailable, + body: map[string]any{"error": "unavailable"}, + want: "the heartbeat could not reach the execution right now; the wait is unchanged, so retry with backoff", + }, + { + name: "rate limited", args: []string{"succeed", token, "--result", "1"}, status: http.StatusTooManyRequests, + body: map[string]any{"error": "too many callback requests"}, + want: "failed to succeed the callback: HTTP 429: too many callback requests", + }, + } { + t.Run(tc.name, func(t *testing.T) { + setCallbacksTestHome(t) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + respondJSON(t, w, tc.status, tc.body) + })) + defer server.Close() + + _, err := executeCommand(t, newCallbacksCommand(server), "", tc.args...) + require.Error(t, err) + assert.Contains(t, err.Error(), tc.want) + assert.Equal(t, tc.status, api.Status(err), "the exit code and telemetry still see the API status") + }) + } +} + +func newCallbacksCommand(server *httptest.Server) *cobra.Command { + return New(cliruntime.Deps{ + CommandPathPrefix: "volcano cloud", + HTTPClient: server.Client(), + APIBaseURL: server.URL, + }) +} + +func executeCommand(t *testing.T, cmd *cobra.Command, stdin string, args ...string) (string, error) { + t.Helper() + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(&out) + cmd.SetIn(bytes.NewBufferString(stdin)) + cmd.SetArgs(args) + err := cmd.Execute() + return out.String(), err +} + +func setCallbacksTestHome(t *testing.T) { + t.Helper() + t.Setenv("HOME", t.TempDir()) + t.Setenv("VOLCANO_TOKEN", "") + t.Setenv("VOLCANO_PROJECT_ID", "") + t.Setenv("VOLCANO_API_URL", "") + t.Setenv("VOLCANO_FIRST_PARTY_DEVICE_CLIENT_ID", "") + + cfg := &cliconfig.Config{ + UserToken: "token", + CurrentProject: &cliconfig.ProjectConfig{ID: projectID, Name: "Beta"}, + } + require.NoError(t, cfg.Save()) +} + +func respondJSON(t *testing.T, w http.ResponseWriter, status int, value any) { + t.Helper() + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + require.NoError(t, json.NewEncoder(w).Encode(value)) +} diff --git a/internal/cmd/durable/callbacks/fail.go b/internal/cmd/durable/callbacks/fail.go new file mode 100644 index 00000000..467ddbaf --- /dev/null +++ b/internal/cmd/durable/callbacks/fail.go @@ -0,0 +1,75 @@ +package callbacks + +import ( + "context" + "errors" + "io" + "strings" + + "github.com/spf13/cobra" + + clidurable "github.com/Kong/volcano-cli/internal/durable" + cliruntime "github.com/Kong/volcano-cli/internal/runtime" +) + +type failOptions struct { + deps cliruntime.Deps + token string + errorType string + message string + jsonOutput bool + out io.Writer +} + +func newFail(deps cliruntime.Deps) *cobra.Command { + opts := failOptions{} + cmd := &cobra.Command{ + Use: "fail ", + Short: "Resume a waiting step with an error", + Long: `Fail one callback. The step waiting on the token resumes by raising +CallbackFailedError, carrying --type and --message. The workflow can catch it +and branch on the type; left uncaught, it fails the execution. + +--type is required: a name the workflow can branch on, such as CardDeclined, up +to 256 characters without control characters. --message is a human-readable +explanation, up to 32 KiB. + +The command returns once the execution has the error, or says Volcano is still +delivering it, as succeed does. + +--json prints the outcome as the API returns it.`, + Example: " " + cliruntime.CommandPath(deps, + `durable callbacks fail vcb_0000000000000000000000000000000000000000000 --type CardDeclined --message "Insufficient funds"`), + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if strings.TrimSpace(opts.errorType) == "" { + return errors.New("--type needs a name the workflow can branch on") + } + token, err := readToken(args[0], cmd.InOrStdin()) + if err != nil { + return err + } + opts.deps = deps + opts.token = token + opts.out = cmd.OutOrStdout() + return runFail(cmd.Context(), opts) + }, + } + cmd.Flags().StringVar(&opts.errorType, "type", "", "Error type the workflow branches on") + cmd.Flags().StringVar(&opts.message, "message", "", "Human-readable explanation of the error") + cmd.Flags().BoolVar(&opts.jsonOutput, "json", false, "Emit machine-readable JSON") + _ = cmd.MarkFlagRequired("type") + return cmd +} + +func runFail(ctx context.Context, opts failOptions) error { + outcome, err := clidurable.NewService(opts.deps).FailCallback(ctx, opts.token, opts.errorType, opts.message) + if err != nil { + return err + } + if opts.jsonOutput { + return writeJSON(opts.out, outcome) + } + reportAnswer(opts.out, outcome, "Failed the callback; the execution resumes with the error") + return nil +} diff --git a/internal/cmd/durable/callbacks/heartbeat.go b/internal/cmd/durable/callbacks/heartbeat.go new file mode 100644 index 00000000..b4557d97 --- /dev/null +++ b/internal/cmd/durable/callbacks/heartbeat.go @@ -0,0 +1,65 @@ +package callbacks + +import ( + "context" + "io" + + "github.com/spf13/cobra" + + clidurable "github.com/Kong/volcano-cli/internal/durable" + "github.com/Kong/volcano-cli/internal/output" + cliruntime "github.com/Kong/volcano-cli/internal/runtime" +) + +type heartbeatOptions struct { + deps cliruntime.Deps + token string + jsonOutput bool + out io.Writer +} + +func newHeartbeat(deps cliruntime.Deps) *cobra.Command { + opts := heartbeatOptions{} + cmd := &cobra.Command{ + Use: "heartbeat ", + Short: "Report that a callback's work is still going", + Long: `Heartbeat one callback, restarting the heartbeat timeout of the step waiting +on the token. A heartbeat never extends the step's own timeout, and a wait +without a heartbeat timeout accepts heartbeats and ignores them. + +Heartbeat more often than the wait's heartbeat timeout. When the heartbeats +stop, the wait resolves with null once the heartbeat timeout passes. + +A heartbeat after the wait ended is refused like a late answer. One that cannot +reach the execution right now leaves the wait unchanged; retry it with backoff. + +--json prints the outcome as the API returns it.`, + Example: " " + cliruntime.CommandPath(deps, + "durable callbacks heartbeat vcb_0000000000000000000000000000000000000000000"), + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + token, err := readToken(args[0], cmd.InOrStdin()) + if err != nil { + return err + } + opts.deps = deps + opts.token = token + opts.out = cmd.OutOrStdout() + return runHeartbeat(cmd.Context(), opts) + }, + } + cmd.Flags().BoolVar(&opts.jsonOutput, "json", false, "Emit machine-readable JSON") + return cmd +} + +func runHeartbeat(ctx context.Context, opts heartbeatOptions) error { + outcome, err := clidurable.NewService(opts.deps).HeartbeatCallback(ctx, opts.token) + if err != nil { + return err + } + if opts.jsonOutput { + return writeJSON(opts.out, outcome) + } + output.Success(opts.out, "Heartbeat sent; the callback is still waiting") + return nil +} diff --git a/internal/cmd/durable/callbacks/succeed.go b/internal/cmd/durable/callbacks/succeed.go new file mode 100644 index 00000000..d6a7cceb --- /dev/null +++ b/internal/cmd/durable/callbacks/succeed.go @@ -0,0 +1,113 @@ +package callbacks + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + + "github.com/spf13/cobra" + + "github.com/Kong/volcano-cli/internal/cmd/cmdutil" + clidurable "github.com/Kong/volcano-cli/internal/durable" + cliruntime "github.com/Kong/volcano-cli/internal/runtime" +) + +type succeedOptions struct { + deps cliruntime.Deps + token string + result json.RawMessage + jsonOutput bool + out io.Writer +} + +func newSucceed(deps cliruntime.Deps) *cobra.Command { + opts := succeedOptions{} + var result, resultFile string + cmd := &cobra.Command{ + Use: "succeed ", + Short: "Resume a waiting step with a result", + Long: `Succeed one callback. The step waiting on the token resumes, and +ctx.waitForCallback returns the result. + +--result is the result's JSON value, and --result-file reads it from a file. +One of them is required. The result can be any JSON value other than null, up +to 64 KiB: a wait that times out resolves with null, so null is refused. + +The command returns once the execution has the result. If handing it over takes +longer than the request waits, the command says Volcano is still delivering it; +Volcano finishes on its own, so do not send it again. + +--json prints the outcome as the API returns it.`, + Example: fmt.Sprintf(` %s + %s`, + cliruntime.CommandPath(deps, + `durable callbacks succeed vcb_0000000000000000000000000000000000000000000 --result '{"charge_id":"ch_1"}'`), + "printf '%s' \"$CALLBACK_TOKEN\" | "+cliruntime.CommandPath(deps, + "durable callbacks succeed - --result-file charge.json")), + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var err error + opts.result, err = resultValue(cmd, result, resultFile) + if err != nil { + return err + } + opts.token, err = readToken(args[0], cmd.InOrStdin()) + if err != nil { + return err + } + opts.deps = deps + opts.out = cmd.OutOrStdout() + return runSucceed(cmd.Context(), opts) + }, + } + cmd.Flags().StringVar(&result, "result", "", "The result's JSON value") + cmd.Flags().StringVar(&resultFile, "result-file", "", "Path to a file holding the result's JSON value") + cmd.Flags().BoolVar(&opts.jsonOutput, "json", false, "Emit machine-readable JSON") + cmd.MarkFlagsMutuallyExclusive("result", "result-file") + cmd.MarkFlagsOneRequired("result", "result-file") + return cmd +} + +// resultValue reads the flag that carries the result and refuses null here, +// where the message can say why. +func resultValue(cmd *cobra.Command, result, resultFile string) (json.RawMessage, error) { + var value json.RawMessage + var err error + if cmd.Flags().Changed("result") { + value, err = cmdutil.ParseJSONValue("result", []byte(result)) + } else { + if strings.TrimSpace(resultFile) == "" { + return nil, errors.New("--result-file needs a path") + } + var contents []byte + contents, err = cmdutil.ReadJSONFile("result", resultFile) + if err != nil { + return nil, err + } + value, err = cmdutil.ParseJSONValue("result", contents) + } + if err != nil { + return nil, err + } + if bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return nil, errors.New("result cannot be null: a wait that times out resolves with null, " + + "so the workflow could not tell the two apart") + } + return value, nil +} + +func runSucceed(ctx context.Context, opts succeedOptions) error { + outcome, err := clidurable.NewService(opts.deps).SucceedCallback(ctx, opts.token, opts.result) + if err != nil { + return err + } + if opts.jsonOutput { + return writeJSON(opts.out, outcome) + } + reportAnswer(opts.out, outcome, "Succeeded the callback; the execution resumes with the result") + return nil +} diff --git a/internal/cmd/durable/durable.go b/internal/cmd/durable/durable.go index 47dfedf5..4f24c208 100644 --- a/internal/cmd/durable/durable.go +++ b/internal/cmd/durable/durable.go @@ -5,6 +5,7 @@ import ( "github.com/spf13/cobra" approvalscmd "github.com/Kong/volcano-cli/internal/cmd/durable/approvals" + callbackscmd "github.com/Kong/volcano-cli/internal/cmd/durable/callbacks" executionscmd "github.com/Kong/volcano-cli/internal/cmd/durable/executions" messagescmd "github.com/Kong/volcano-cli/internal/cmd/durable/messages" schedulerscmd "github.com/Kong/volcano-cli/internal/cmd/durable/schedulers" @@ -43,5 +44,6 @@ collections never accept each other's names or ids.`, cmd.AddCommand(schedulerscmd.New(deps)) cmd.AddCommand(approvalscmd.New(deps)) cmd.AddCommand(messagescmd.New(deps)) + cmd.AddCommand(callbackscmd.New(deps)) return cmd } diff --git a/internal/durable/callbacks.go b/internal/durable/callbacks.go new file mode 100644 index 00000000..fa8dfcc3 --- /dev/null +++ b/internal/durable/callbacks.go @@ -0,0 +1,118 @@ +package durable + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "regexp" + + "github.com/Kong/volcano-cli/internal/api" + "github.com/Kong/volcano-cli/internal/apiclient" +) + +const maxCallbackTokenLength = 128 + +// callbackToken is the shape the API accepts. A token is checked here so a +// malformed one is never sent: a validation error could quote it back, and +// error text reaches usage telemetry. +var callbackToken = regexp.MustCompile(`^vcb_[A-Za-z0-9_-]+$`) + +// ValidateCallbackToken refuses a value that cannot be a callback token. The +// error never quotes the value, which may be a token with a typo in it. +func ValidateCallbackToken(token string) error { + if len(token) > maxCallbackTokenLength || !callbackToken.MatchString(token) { + return fmt.Errorf("invalid callback token: a token is vcb_ followed by letters, digits, '-' or '_', "+ + "up to %d characters in all", maxCallbackTokenLength) + } + return nil +} + +// SucceedCallback resumes the step waiting on token with result. +func (s Service) SucceedCallback( + ctx context.Context, token string, result json.RawMessage, +) (*apiclient.DurableCallbackOutcome, error) { + client, err := s.callbackClient(token) + if err != nil { + return nil, err + } + outcome, err := client.SucceedDurableCallback(ctx, token, result) + if err != nil { + return nil, callbackError("succeed", err) + } + return outcome, nil +} + +// FailCallback resumes the step waiting on token by raising an error with +// errorType and message. +func (s Service) FailCallback( + ctx context.Context, token, errorType, message string, +) (*apiclient.DurableCallbackOutcome, error) { + client, err := s.callbackClient(token) + if err != nil { + return nil, err + } + outcome, err := client.FailDurableCallback(ctx, token, errorType, message) + if err != nil { + return nil, callbackError("fail", err) + } + return outcome, nil +} + +// HeartbeatCallback restarts the heartbeat timeout of the step waiting on +// token. +func (s Service) HeartbeatCallback(ctx context.Context, token string) (*apiclient.DurableCallbackOutcome, error) { + client, err := s.callbackClient(token) + if err != nil { + return nil, err + } + outcome, err := client.HeartbeatDurableCallback(ctx, token) + if err != nil { + if api.Status(err) == http.StatusServiceUnavailable { + return nil, api.WithCause(errors.New("the heartbeat could not reach the execution right now; "+ + "the wait is unchanged, so retry with backoff"), err) + } + return nil, callbackError("heartbeat", err) + } + return outcome, nil +} + +// callbackClient sends no credential: the token is the only one the callback +// routes take, so answering a callback needs no login or project. That is what +// lets the system a workflow handed the token to run these commands. +func (s Service) callbackClient(token string) (*api.Client, error) { + if err := ValidateCallbackToken(token); err != nil { + return nil, err + } + cfg, err := s.sessions.Config() + if err != nil { + return nil, err + } + client, err := s.sessions.APIClient(s.sessions.APIURL(cfg), "") + if err != nil { + return nil, fmt.Errorf("failed to create api client: %w", err) + } + return client, nil +} + +// callbackError says what became of the wait when the API refuses an answer, +// because a holder decides from it whether to retry. The API error stays the +// cause, so the exit code still reflects its status. +func callbackError(action string, err error) error { + var message string + switch api.Code(err) { + case "callback_not_found": + message = "no callback has this token; its execution may be past its retention" + case "callback_completed": + message = "the callback already has an outcome; a token takes one answer, " + + "so a retry of an answer that went through ends here too" + case "callback_expired": + message = "the callback timed out; nothing will take an answer, so do not retry" + case "execution_ended": + message = "the callback's execution has ended; nothing will take an answer, so do not retry" + default: + return fmt.Errorf("failed to %s the callback: %w", action, err) + } + return api.WithCause(errors.New(message), err) +} diff --git a/internal/telemetry/fields.go b/internal/telemetry/fields.go index 45664741..a9355ade 100644 --- a/internal/telemetry/fields.go +++ b/internal/telemetry/fields.go @@ -177,7 +177,7 @@ const redacted = "[redacted]" var ( bearerPattern = regexp.MustCompile(`(?i)\bbearer\s+\S+`) jwtPattern = regexp.MustCompile(`eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]*`) - keyPattern = regexp.MustCompile(`\b(?:(?:(?:pk|pt|sk|ak)-|vbt_|vpg_)[A-Za-z0-9_-]{16,}|sbp1\.[A-Za-z0-9+/=:]{16,})`) + keyPattern = regexp.MustCompile(`\b(?:(?:(?:pk|pt|sk|ak)-|vbt_|vpg_|vcb_)[A-Za-z0-9_-]{16,}|sbp1\.[A-Za-z0-9+/=:]{16,})`) userPattern = regexp.MustCompile(`([a-zA-Z][a-zA-Z0-9+.-]*://)[^\s/?#]*@`) emailPattern = regexp.MustCompile(`[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)*\.[A-Za-z]{2,}`) queryPattern = regexp.MustCompile(`(https?://[^\s?#]+)\?[^\s#]*`) diff --git a/internal/telemetry/fields_test.go b/internal/telemetry/fields_test.go index 888ca545..a5706e6a 100644 --- a/internal/telemetry/fields_test.go +++ b/internal/telemetry/fields_test.go @@ -107,6 +107,7 @@ func TestRedact(t *testing.T) { "proxy sbp1.AbC+dEf/GhI0123456789= denied": "proxy [redacted] denied", "exchange vbt_0123456789abcdefABCD failed": "exchange [redacted] failed", "login vpg_0123456789abcdef_ABCD-xyz refused": "login [redacted] refused", + "answer vcb_0123456789abcdef_ABCD-xyz refused": "answer [redacted] refused", "GET https://x.dev/o/a.png?X-Amz-Signature=abc&b=1 failed": "GET https://x.dev/o/a.png?[redacted] failed", "GET https://example.com?email=a@b.com&key=x failed": "GET https://example.com?[redacted] failed", "dial postgres://admin:p@ss@db:5432/app failed": "dial postgres://[redacted]@db:5432/app failed", diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index d959b52d..80bb1e59 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -6645,6 +6645,232 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + /durable-callbacks/succeed: + post: + tags: + - Durable Functions + summary: Complete a waiting step with a result + description: | + Resumes the workflow step waiting on the token with `result`. Call it + from the system the workflow handed the token to. + + The token is the only credential. A token takes one outcome: once it + has succeeded or failed, every later request is refused with + `409 callback_completed`, including a retry of the request that + succeeded it. + operationId: succeedDurableCallback + security: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/SucceedDurableCallbackRequest' + responses: + '200': + description: The workflow has the result + content: + application/json: + schema: + $ref: '#/components/schemas/DurableCallbackOutcome' + '202': + description: | + The result is recorded and on its way to the workflow, which takes + longer than this request waits. Volcano finishes the delivery; do + not send it again. + content: + application/json: + schema: + $ref: '#/components/schemas/DurableCallbackOutcome' + '400': + description: The token is missing, or `result` is missing, null, not JSON, or larger than 64 KiB + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: | + No callback has this token. `code` is `callback_not_found`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '409': + description: | + The token already has an outcome. `code` is `callback_completed`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '410': + description: | + The workflow no longer waits on the token. `code` says why: + + - `callback_expired`: the step timed out, on its own timeout or + its heartbeat timeout. + - `execution_ended`: the execution finished or stopped. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '413': + description: The request body is larger than 72 KiB + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '429': + description: Too many callback requests for the token or its project. Retry later. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + /durable-callbacks/fail: + post: + tags: + - Durable Functions + summary: Fail a waiting step + description: | + Resumes the workflow step waiting on the token by raising `error`. The + SDKs raise it as `CallbackFailedError`, carrying its `type` and + `message`. + + The token is the only credential, and takes one outcome, as for + succeeding it. + operationId: failDurableCallback + security: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/FailDurableCallbackRequest' + responses: + '200': + description: The workflow has the error + content: + application/json: + schema: + $ref: '#/components/schemas/DurableCallbackOutcome' + '202': + description: | + The error is recorded and on its way to the workflow. Volcano + finishes the delivery; do not send it again. + content: + application/json: + schema: + $ref: '#/components/schemas/DurableCallbackOutcome' + '400': + description: The token is missing, or the error's `type` or `message` is invalid + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: | + No callback has this token. `code` is `callback_not_found`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '409': + description: | + The token already has an outcome. `code` is `callback_completed`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '410': + description: | + The workflow no longer waits on the token. `code` is + `callback_expired` or `execution_ended`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '413': + description: The request body is larger than 200 KiB + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '429': + description: Too many callback requests for the token or its project. Retry later. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + /durable-callbacks/heartbeat: + post: + tags: + - Durable Functions + summary: Keep a waiting step alive + description: | + Tells the workflow step waiting on the token that the work is still + going on, restarting its heartbeat timeout. A heartbeat never extends + the step's own timeout, and a step without a heartbeat timeout needs + none. + operationId: heartbeatDurableCallback + security: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/HeartbeatDurableCallbackRequest' + responses: + '200': + description: The step is still waiting + content: + application/json: + schema: + $ref: '#/components/schemas/DurableCallbackOutcome' + '400': + description: The token is missing + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: | + No callback has this token. `code` is `callback_not_found`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '409': + description: | + The token already has an outcome. `code` is `callback_completed`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '410': + description: | + The workflow no longer waits on the token. `code` is + `callback_expired` or `execution_ended`. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '413': + description: The request body is larger than 8 KiB + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '429': + description: Too many callback requests for the token or its project. Retry later. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '503': + description: The heartbeat could not reach the workflow right now. Retry with backoff. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' /durable-approvals: post: tags: @@ -23391,6 +23617,92 @@ components: - limit - total - has_more + SucceedDurableCallbackRequest: + type: object + properties: + token: + $ref: '#/components/schemas/DurableCallbackToken' + result: + x-go-type: json.RawMessage + x-go-type-skip-optional-pointer: true + description: | + Any JSON value other than null, up to 64 KiB. The waiting step + resolves with it. Null is refused because a timeout resolves the + step with null. + required: + - token + - result + additionalProperties: false + FailDurableCallbackRequest: + type: object + properties: + token: + $ref: '#/components/schemas/DurableCallbackToken' + error: + $ref: '#/components/schemas/DurableCallbackError' + required: + - token + - error + additionalProperties: false + HeartbeatDurableCallbackRequest: + type: object + properties: + token: + $ref: '#/components/schemas/DurableCallbackToken' + required: + - token + additionalProperties: false + DurableCallbackOutcome: + type: object + properties: + status: + $ref: '#/components/schemas/DurableCallbackStatus' + required: + - status + DurableCallbackToken: + type: string + minLength: 5 + maxLength: 128 + pattern: ^vcb_[A-Za-z0-9_-]+$ + description: | + The `vcb_` token the workflow handed out. It is the only credential + the callback operations take, so treat it as a secret. + DurableCallbackStatus: + type: string + enum: + - waiting + - delivering + - succeeded + - failed + description: | + `waiting` until the token is succeeded or failed. `delivering` while + the outcome is on its way to the workflow; Volcano finishes the + delivery on its own. `succeeded` or `failed` once the workflow has the + outcome. + DurableCallbackError: + type: object + description: The error the waiting step raises. + properties: + type: + type: string + minLength: 1 + maxLength: 256 + pattern: ^[^\x00-\x1F\x7F]+$ + description: | + A name workflows can branch on, such as `PaymentDeclined`. Up to + 256 characters, without control characters. + message: + type: string + maxLength: 32768 + pattern: ^[^\x00-\x08\x0B-\x1F\x7F]*$ + description: | + A human-readable explanation of up to 32,768 bytes of UTF-8, with + tab and newline as its only control characters. A character + outside ASCII takes more than one byte, so such text reaches the + limit before `maxLength`. + required: + - type + additionalProperties: false DurableApproval: type: object description: An approval a durable workflow requested. diff --git a/tests/e2e/api/durable_callbacks_test.go b/tests/e2e/api/durable_callbacks_test.go new file mode 100644 index 00000000..922ea2fd --- /dev/null +++ b/tests/e2e/api/durable_callbacks_test.go @@ -0,0 +1,146 @@ +package api + +import ( + "path/filepath" + "regexp" + "strings" + "testing" + "time" +) + +// apiE2EDurableCallbackSource waits on two callbacks: payment is succeeded and +// refund is failed. Each submitter logs its token, which the test reads back +// from the runtime logs to play the system the token was handed to. A real +// workflow sends it somewhere instead: a token is a secret. +const apiE2EDurableCallbackSource = ` +const { durable, CallbackFailedError } = require('@volcano.dev/sdk/durable'); + +const handOff = (name) => (token) => { + console.log('callback-token ' + name + ' ' + token); +}; + +exports.handler = durable(async (input, ctx) => { + const paid = await ctx.waitForCallback('payment', handOff('payment'), { + timeout: '10m', + heartbeatTimeout: '5m', + }); + let declined = null; + try { + await ctx.waitForCallback('refund', handOff('refund'), { timeout: '10m' }); + } catch (error) { + if (!(error instanceof CallbackFailedError)) throw error; + declined = { type: error.type, message: error.message }; + } + return { paid, declined }; +}); +` + +// apiE2EDurableCallbackManifest pins the first SDK release that ships +// ctx.waitForCallback. +const apiE2EDurableCallbackManifest = `{ + "name": "callback-pipeline", + "private": true, + "dependencies": { + "@volcano.dev/sdk": "^2.2.0" + } +} +` + +// apiE2EUnknownCallbackToken is well formed and was never issued. +const apiE2EUnknownCallbackToken = "vcb_0000000000000000000000000000000000000000000" + +// TestAPIE2ECloudDurableCallbacks walks the callbacks group against real +// infrastructure: a heartbeat and a success on one token, a failure on the +// next, each followed into the execution that resumes with it. +// +// The answers run without a login, the way the system a workflow hands a token +// to would run them. +func TestAPIE2ECloudDurableCallbacks(t *testing.T) { + env := setupAPIE2E(t, "cloud-callbacks") + writeAPIE2EDurableCallbackProject(t, env.projectDir) + + env.loginAndUse(t) + env.runCloudCLI(t, "durable", "deploy", "--all").requireSuccess(t, + "[1/1] Deploying callback-pipeline", "1/1 durable function(s) deployment started") + t.Cleanup(func() { + env.runCloudCLI(t, "durable", "delete", "callback-pipeline", "--yes") + }) + env.waitForCloudCLIContains(t, apiE2EFunctionDeploymentTimeout, "Status: active", + "durable", "get", "callback-pipeline") + + start := env.runCloudCLI(t, "durable", "start", "callback-pipeline", "--name", "cli-e2e-callbacks") + start.requireSuccess(t, "Execution cli-e2e-callbacks started") + executionID := apiE2EDurableExecutionID(t, start.output) + + holder := []string{"HOME=" + t.TempDir()} + payment := waitForAPIE2ECallbackToken(t, env, "payment") + env.runCloudCLIWithEnv(t, holder, "durable", "callbacks", "heartbeat", payment). + requireSuccess(t, "Heartbeat sent; the callback is still waiting") + requireAPIE2ECallbackAccepted(t, env.runCloudCLIWithEnv(t, holder, "durable", "callbacks", "succeed", payment, + "--result", `{"charge_id":"ch_4417"}`, "--json"), "succeeded") + env.runCloudCLIWithEnv(t, holder, "durable", "callbacks", "succeed", payment, "--result", "1"). + requireFailure(t, "the callback already has an outcome") + env.runCloudCLIWithEnv(t, holder, "durable", "callbacks", "heartbeat", payment). + requireFailure(t, "the callback already has an outcome") + + refund := waitForAPIE2ECallbackToken(t, env, "refund") + requireAPIE2ECallbackAccepted(t, env.runCloudCLIWithEnv(t, holder, "durable", "callbacks", "fail", refund, + "--type", "CardDeclined", "--message", "Insufficient funds", "--json"), "failed") + + env.waitForCloudCLIContains(t, apiE2EDurableExecutionTimeout, "Status: succeeded", + "durable", "executions", "get", "callback-pipeline", executionID) + env.runCloudCLI(t, "durable", "executions", "get", "callback-pipeline", executionID). + requireSuccess(t, `"charge_id": "ch_4417"`, `"type": "CardDeclined"`, `"message": "Insufficient funds"`) + + env.runCloudCLIWithEnv(t, holder, "durable", "callbacks", "heartbeat", apiE2EUnknownCallbackToken). + requireFailure(t, "no callback has this token") + env.runCloudCLIWithEnv(t, holder, "durable", "callbacks", "heartbeat", "vcb_not/a/token"). + requireFailure(t, "invalid callback token") +} + +// requireAPIE2ECallbackAccepted passes an answer the execution has, or one +// Volcano recorded and is still delivering. The execution's own status is what +// shows the answer arrived. +func requireAPIE2ECallbackAccepted(t *testing.T, answer cliResult, status string) { + t.Helper() + answer.requireSuccess(t) + if !strings.Contains(answer.output, `"status": "`+status+`"`) && + !strings.Contains(answer.output, `"status": "delivering"`) { + t.Fatalf("the answer was neither %s nor delivering:\n%s", status, redactCredentials(answer.output)) + } +} + +// waitForAPIE2ECallbackToken polls the runtime logs until the submitter of the +// named callback has logged its token. The submitter runs once the wait opens, +// after the start returns, and the log takes a while to be searchable. +func waitForAPIE2ECallbackToken(t *testing.T, env *apiE2E, name string) string { + t.Helper() + logged := regexp.MustCompile(`callback-token ` + regexp.QuoteMeta(name) + ` (vcb_[A-Za-z0-9_-]+)`) + deadline := time.Now().Add(apiE2EDurableExecutionTimeout) + var last cliResult + for time.Now().Before(deadline) { + last = env.runCloudCLIWithin(t, attemptTimeout(deadline), + "durable", "logs", "callback-pipeline", "--type", "runtime") + if match := logged.FindStringSubmatch(last.output); last.code == 0 && match != nil { + return match[1] + } + time.Sleep(apiE2EPollInterval) + } + t.Fatalf("the %s callback never logged its token:\n%s", name, redactCredentials(last.output)) + return "" +} + +func writeAPIE2EDurableCallbackProject(t *testing.T, projectDir string) { + t.Helper() + writeAPIE2EBaseProject(t, projectDir) + functionDir := filepath.Join(projectDir, "volcano", "functions", "callback-pipeline") + writeAPIE2EFile(t, filepath.Join(functionDir, "index.js"), apiE2EDurableCallbackSource) + writeAPIE2EFile(t, filepath.Join(functionDir, "package.json"), apiE2EDurableCallbackManifest) + writeAPIE2EFile(t, filepath.Join(projectDir, "volcano-config.yaml"), ` +version: 1 +functions: + - name: hello + - name: callback-pipeline + kind: durable +`) +} diff --git a/tests/e2e/localmode/durable_callbacks_test.go b/tests/e2e/localmode/durable_callbacks_test.go new file mode 100644 index 00000000..14ff81d2 --- /dev/null +++ b/tests/e2e/localmode/durable_callbacks_test.go @@ -0,0 +1,136 @@ +package localmode + +import ( + "os" + "path/filepath" + "regexp" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// localModeE2EDurableCallbackSource is the cloud E2E's workflow in Python: +// callbacks work the same in both modes, and so does the code a user writes for +// them. Each submitter prints its token for the test to answer with. +const localModeE2EDurableCallbackSource = ` +from volcano_sdk.durable_authoring import CallbackFailedError, durable + + +def hand_off(name): + def submit(token): + print(f"callback-token {name} {token}", flush=True) + + return submit + + +@durable +def handler(event, ctx): + paid = ctx.wait_for_callback("payment", hand_off("payment"), timeout="10m", heartbeat_timeout="5m") + declined = None + try: + ctx.wait_for_callback("refund", hand_off("refund"), timeout="10m") + except CallbackFailedError as error: + declined = {"type": error.type, "message": error.message} + return {"paid": paid, "declined": declined} +` + +// localModeE2EDurableCallbackRequirements pins the first volcano-sdk-python +// release with ctx.wait_for_callback. +const localModeE2EDurableCallbackRequirements = "volcano-sdk-python~=0.16.0\n" + +// TestLocalModeE2EDurableCallbacks is the cloud callbacks round trip against +// the local stack: a heartbeat and a success on one token, a failure on the +// next, and the execution resuming with both. It has its own stack and +// project, so a failure here does not stop the smoke's later checks. +func TestLocalModeE2EDurableCallbacks(t *testing.T) { + if os.Getenv("VOLCANO_LOCALMODE_E2E") != "1" { + t.Skip("set VOLCANO_LOCALMODE_E2E=1 to run disposable Docker acceptance") + } + requireDocker(t) + binary := buildLocalModeE2EBinary(t) + env, dir := localModeE2EEnv(t), localModeE2EProjectDir(t) + cleanup := func() { + _, _ = runVolcanoLocalModeE2EAllowFailure(t, binary, env, dir, "stop", "--clean") + } + cleanup() + t.Cleanup(cleanup) + + functionDir := filepath.Join("volcano", "functions", "callback-pipeline") + writeLocalModeE2EFile(t, dir, "volcano-config.yaml", ` +version: 1 +functions: + - name: callback-pipeline + kind: durable +`) + writeLocalModeE2EFile(t, dir, filepath.Join(functionDir, "main.py"), localModeE2EDurableCallbackSource) + writeLocalModeE2EFile(t, dir, filepath.Join(functionDir, "requirements.txt"), localModeE2EDurableCallbackRequirements) + requireContains(t, runVolcanoLocalModeE2E(t, binary, env, dir, "start"), + "Volcano is ready for local development.") + + requireContains(t, runVolcanoLocalModeE2E(t, binary, env, dir, "durable", "deploy", "--all"), + "callback-pipeline") + waitForVolcanoLocalModeE2EContains(t, binary, env, dir, "Status: active", + "durable", "get", "callback-pipeline") + + started := runVolcanoLocalModeE2E(t, binary, env, dir, + "durable", "start", "callback-pipeline", "--name", "cli-local-callbacks") + executionID := localModeE2EFieldValue(t, started, "ID") + + payment := waitForLocalModeE2ECallbackToken(t, binary, env, dir, "payment") + requireContains(t, runVolcanoLocalModeE2E(t, binary, env, dir, + "durable", "callbacks", "heartbeat", payment), "Heartbeat sent; the callback is still waiting") + requireLocalModeE2ECallbackAccepted(t, runVolcanoLocalModeE2E(t, binary, env, dir, + "durable", "callbacks", "succeed", payment, "--result", `{"charge_id":"ch_4417"}`, "--json"), "succeeded") + answered, err := runVolcanoLocalModeE2EAllowFailure(t, binary, env, dir, + "durable", "callbacks", "succeed", payment, "--result", "1") + require.Error(t, err, "a token takes one answer\n%s", answered) + requireContains(t, answered, "the callback already has an outcome") + + refund := waitForLocalModeE2ECallbackToken(t, binary, env, dir, "refund") + requireLocalModeE2ECallbackAccepted(t, runVolcanoLocalModeE2E(t, binary, env, dir, + "durable", "callbacks", "fail", refund, "--type", "CardDeclined", "--message", "Insufficient funds", "--json"), + "failed") + + execution := awaitLocalModeE2EDurableExecution(t, binary, env, dir, "callback-pipeline", executionID) + requireContains(t, execution, "Status: succeeded") + requireContains(t, execution, `"charge_id": "ch_4417"`) + requireContains(t, execution, `"type": "CardDeclined"`) + + unknown, err := runVolcanoLocalModeE2EAllowFailure(t, binary, env, dir, + "durable", "callbacks", "heartbeat", "vcb_0000000000000000000000000000000000000000000") + require.Error(t, err, "a token nobody issued has to be refused\n%s", unknown) + requireContains(t, unknown, "no callback has this token") +} + +// requireLocalModeE2ECallbackAccepted passes an answer the execution has, or +// one Volcano recorded and is still delivering. The execution's own status is +// what shows the answer arrived. +func requireLocalModeE2ECallbackAccepted(t *testing.T, answer, status string) { + t.Helper() + if !strings.Contains(answer, `"status": "`+status+`"`) && !strings.Contains(answer, `"status": "delivering"`) { + t.Fatalf("the answer was neither %s nor delivering:\n%s", status, answer) + } +} + +// waitForLocalModeE2ECallbackToken polls the runtime logs until the submitter +// of the named callback has logged its token. +func waitForLocalModeE2ECallbackToken(t *testing.T, binary string, env []string, dir, name string) string { + t.Helper() + + logged := regexp.MustCompile(`callback-token ` + regexp.QuoteMeta(name) + ` (vcb_[A-Za-z0-9_-]+)`) + deadline := time.Now().Add(3 * time.Minute) + var last string + for time.Now().Before(deadline) { + var err error + last, err = runVolcanoLocalModeE2EAllowFailure(t, binary, env, dir, + "durable", "logs", "callback-pipeline", "--type", "runtime") + if match := logged.FindStringSubmatch(last); err == nil && match != nil { + return match[1] + } + time.Sleep(2 * time.Second) + } + t.Fatalf("the %s callback never logged its token:\n%s", name, last) + return "" +}