feat(durable): answer durable callbacks with their token

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 ./...

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) |

docs/durable-functions.md

@@ -77,11 +77,14 @@ existing function's scope alone.
 | Send a message | `volcano cloud durable messages send <function> <execution-id> <name> [--data … \| --data-file …] [--idempotency-key …]` |
 | List messages | `volcano cloud durable messages list <function> <execution-id> [--direction in\|out] [--name …] [--after …] [--limit 100] [--follow]` |
 | Open message waits | `volcano cloud durable messages waits <function> <execution-id>` |
+| Succeed a callback | `volcano cloud durable callbacks succeed <token> --result … \| --result-file …` |
+| Fail a callback | `volcano cloud durable callbacks fail <token> --type … [--message …]` |
+| Heartbeat a callback | `volcano cloud durable callbacks heartbeat <token>` |
 
 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

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

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
+}

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":"<b>A & B</b>"}`))
+			},
+			want: `"note":"<b>A & B</b>"`,
+		},
+		{
+			name: "fail",
+			answer: func(c *Client) (*apiclient.DurableCallbackOutcome, error) {
+				return c.FailDurableCallback(context.Background(), durableTestCallbackToken,
+					"Rejected", "<b>A & B</b>")
+			},
+			want: `"message":"<b>A & B</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))
+		})
+	}
+}