diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6799cff..28e4057 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -222,20 +222,22 @@ jobs: # until the release PR merges — and `go get` would fail here for a reason that is not a # bug. Skip that case in ordinary CI; release CI requires the tags to exist. core="$(sed -n 's|^[[:space:]]*github.com/ConfigButler/krm-stream/gateway v\([^ ]*\).*|\1|p' gateway/kube/go.mod | head -1)" - case "$core" in - 0.0.0-*) ;; # a pseudo-version names a commit, and always resolves - "") - echo "::error::gateway/kube/go.mod does not require the core module at all" && exit 1 ;; - *) - if ! git ls-remote --exit-code --tags origin "refs/tags/gateway/v${core}" >/dev/null 2>&1; then - if [ -n "$RELEASE_VERSION" ]; then - echo "::error::released adapter requires missing core tag gateway/v${core}" - exit 1 - fi - echo "::notice::core v${core} is not tagged yet; release CI verifies the public tag after tagging." - exit 0 - fi ;; - esac + if [ -z "$core" ]; then + echo "::error::gateway/kube/go.mod does not require the core module at all" && exit 1 + fi + # A pseudo-version names a commit, and always resolves. It ends in a 14-digit timestamp and a + # 12-character revision, whatever its base: v0.0.0-…, or v0.4.1-0.… once a tag precedes it, + # which is the form Go chooses when an adapter change needs unreleased core API. + if ! printf '%s' "$core" | grep -Eq '[-.][0-9]{14}-[0-9a-f]{12}$'; then + if ! git ls-remote --exit-code --tags origin "refs/tags/gateway/v${core}" >/dev/null 2>&1; then + if [ -n "$RELEASE_VERSION" ]; then + echo "::error::released adapter requires missing core tag gateway/v${core}" + exit 1 + fi + echo "::notice::core v${core} is not tagged yet; release CI verifies the public tag after tagging." + exit 0 + fi + fi revision="${{ inputs.ref || github.event.pull_request.head.sha || github.sha }}" if [ -n "$RELEASE_VERSION" ]; then diff --git a/README.md b/README.md index bb77334..13d02e7 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,13 @@ version: see [why a quiet stream can reject a save](docs/saving.md#why-a-quiet-s ## Start here +**Want it running in minutes?** [krm-foyer](https://github.com/ConfigButler/krm-foyer) is a +ready-made backend for browser apps on Kubernetes: OIDC sign-in, server-side sessions, a `/k8s` API +proxy, and krm-stream's live resources, with each watch opened as the signed-in user so Kubernetes +RBAC decides what they see. Its `task demo` starts a test cluster and an example app. Start there if +you don't already have a Go backend to embed the gateway in. To build the integration yourself, read +on. + There are two halves, and they are usually two different people. ### The browser half diff --git a/docs/adopting.md b/docs/adopting.md index de8967f..bdf740c 100644 --- a/docs/adopting.md +++ b/docs/adopting.md @@ -19,6 +19,28 @@ namespaced resource. `ScopePolicy` makes the host choose explicitly. Keep all-namespaces access rare. It changes the size, disclosure risk, and operating cost of a stream. Use an `Authorizer` to pin a user to a namespace or target before any watch opens. +### Delegating resource admission to Kubernetes + +A host whose endpoint should expose whatever Kubernetes lets each user read, as a `/k8s` proxy does, +has no resource list to keep. `AnyResource` says so explicitly: + +```go +Scopes: gateway.ScopePolicy{ + Targets: []string{""}, + AnyResource: true, // admission is left to the upstream's own authorization +}, +``` + +Any resource the scope syntax accepts then passes, with or without a namespace. Without one, the API +server decides: the list of a cluster-scoped resource, or every namespace for a namespaced one, which +RBAC allows only with a cluster-wide grant. The target allowlist, `AllowLabelSelector`, the refusal of +API-server addresses and the `Authorizer` all still apply. + +`AnyResource` is safe only when `Clients` returns a backend built from the caller's own credential. +The library cannot check that: a `kube.Backend` can carry a service account's client, and a wrapper +can hide a `SharedBackend`. With either, every resource that identity can read is open to everyone +your `Authorizer` admits. Prove the binding in the host's own tests. + ### Errors from the exported surface are `error` `ScopeFromQuery` and `ScopePolicy.Validate` return `error`, and the concrete value is a @@ -31,11 +53,28 @@ if errors.As(err, &serr) { } ``` +### Errors on the wire, and the detail behind them + +A `StreamError` you return keeps its `Message`; its `Cause` never reaches the wire. Any other error +reaches the browser as `INTERNAL` with the message `internal error`. Set `Options.Diagnostics` to see +the raw error, with the principal and scope it belongs to, and log what your policy allows: + +```go +Diagnostics: func(d gateway.Diagnostic) { + slog.Warn("stream error", "code", d.Code, "resource", d.Scope.Resource, "err", redact(d.Err)) +}, +``` + +`Principal` may return a `*StreamError` too: `gateway.Unauthenticated` when signing in again will +help, or `gateway.UpstreamUnavailable` when your session store is down. Any other `Principal` error +is sent as `UNAUTHENTICATED` "not authenticated". + ### Targets with a path prefix A `rest.Config.Host` may include a path prefix, such as a kcp workspace URL (`https://kcp.example/clusters/root:org:ws`). The dynamic client preserves it when constructing API -requests. Use `kube.NewBackendForConfig(cfg)` to include the endpoint in upstream error diagnostics; +requests. Use `kube.NewBackendForConfig(cfg)` to include the endpoint in upstream errors, which reach +`Options.Diagnostics` and never the browser; `kube.NewBackend(dynamicClient)` also works when a dynamic client already exists. ## 2. Mount the same-origin cookie endpoint diff --git a/docs/auth.md b/docs/auth.md index e00fd18..0bfc2df 100644 --- a/docs/auth.md +++ b/docs/auth.md @@ -53,10 +53,14 @@ gateway.Handler(gateway.Options{ }) ``` -- `Principal` resolves the request to an opaque application identity. +- `Principal` resolves the request to an opaque application identity. Return a `*StreamError` to + choose the refusal (`UNAUTHENTICATED`, `UPSTREAM_UNAVAILABLE`); any other error is sent as + `UNAUTHENTICATED` without its text. - `Authorizer` denies unauthorized scopes before a watch opens and on subsequent checks. - `Clients` is a `ClientFor` callback supplying the backend for that identity and target. -- `Scopes` allowlists targets and resources. A browser cannot supply a raw API-server URL. +- `Scopes` allowlists targets and resources, or with `AnyResource` leaves resource admission to the + API server's RBAC for a backend that acts as the caller. A browser cannot supply a raw API-server + URL. A per-user backend can use the user's bearer token or Kubernetes impersonation. Impersonation requires explicit host credentials with impersonation rights. Scope and disclosure policy remain @@ -76,7 +80,9 @@ options.ReauthorizationTimeout = 5 * time.Second ``` Timed checks run per subscriber and pause that subscriber's object delivery. Denial, timeout, policy -failure or a changed projection terminates only that stream; other subscribers continue. Zero +failure or a changed projection terminates only that stream; other subscribers continue. A +SubjectAccessReview that cannot reach the API server ends the stream with a non-terminal +`UPSTREAM_UNAVAILABLE`, so the client may reconnect once it is back. Zero interval keeps cycle-only checks; zero timeout uses 10 seconds. The bound assumes callbacks honor context cancellation and sinks do not block indefinitely. diff --git a/docs/client-state-model.md b/docs/client-state-model.md index b1eaf65..6418e84 100644 --- a/docs/client-state-model.md +++ b/docs/client-state-model.md @@ -25,7 +25,9 @@ Use `applyStreamEvent` or the transport helpers instead of translating protocol - `deleted` removes the resource and its draft. - `synced` prunes resources that were not seen during the completed snapshot. - `error` is terminal only when the event says it is terminal. `RESYNC_REQUIRED` starts a new - snapshot on the same connection. + snapshot on the same connection. After any other non-terminal error, such as + `UPSTREAM_UNAVAILABLE`, the gateway closes the connection and the client reconnects, waiting at + least `retryAfterMs`. An incomplete snapshot never prunes state. That prevents a transient disconnect from making a UI lose objects it has not yet reloaded. diff --git a/docs/field-reports/first-fieldreport.md b/docs/field-reports/first-fieldreport.md new file mode 100644 index 0000000..3266fe1 --- /dev/null +++ b/docs/field-reports/first-fieldreport.md @@ -0,0 +1,210 @@ +# krm-stream feedback from krm-foyer + +krm-foyer is about to host the krm-stream gateway with user-authenticated watches +([roadmap step 5](https://github.com/ConfigButler/krm-foyer/blob/main/docs/roadmap.md#order-of-work)). This note asks for five changes, most +important first. None of them blocks us: for each we say what krm-foyer does meanwhile. +Everything was checked against krm-stream v0.4.0 (`gateway`, `gateway/kube` and +`@configbutler/krm-stream`) on 2026-10-02, by reading the source and, for the error +output below, by running the gateway. + +## The asks at a glance + +| # | Ask | Priority | What krm-foyer does meanwhile | +| --- | --- | --- | --- | +| 1 | Map Kubernetes errors to the spec's codes, and recover from the retryable ones with backoff | High | Wraps `kube.Backend` and maps errors itself | +| 2 | Send a generic message with an unexpected `INTERNAL`, and give the detail to a diagnostic hook | High | The same wrapper keeps raw errors from reaching the gateway | +| 3 | Let `Handler` serve a host that delegates resource admission to Kubernetes | Medium | Parses the scope itself and calls `Gateway.ServeStream` | +| 4 | Let `Principal` say why it failed, so a missing credential is `UNAUTHENTICATED` | Low | Answers before the gateway runs, so it never sees this | +| 5 | Honour retry hints at the layer that retries | Low | Nothing needed for now | + +## How krm-foyer uses krm-stream + +krm-foyer is a backend for frontend: it signs users in with OIDC, keeps their ID token +server-side behind a session cookie, and proxies `/k8s` to the API server with that +token. **Kubernetes alone decides what a user may do:** krm-foyer has no allowlist of +its own, and no service-account fallback. Its [design](https://github.com/ConfigButler/krm-foyer/blob/main/docs/design.md#access) promises +that for `/stream` as well as `/k8s`. + +So krm-foyer uses the gateway in per-user mode: + +- **Before the gateway,** krm-foyer checks the session and answers 401 itself when there + is none. It parses the scope with `ScopeFromQuery` and accepts only its one target. +- **`Clients`** returns `kube.NewBackend` over a `rest.Config` holding only the API + server's address, its CA and the user's token. Tests that try to get past this boundary + are krm-foyer's to write; no type check in krm-stream could prove it. +- **`Authorizer`** is `AllowAll`, because every watch is opened as the user, so + Kubernetes RBAC decides. +- **When the session ends,** krm-foyer cancels the request context. That aborts the + stream and cancels the upstream watch; the managed connector then gets a 401 and stops. + +Shared watches come later. Everything below is about per-user mode unless it says +otherwise. + +## Ask 1: map Kubernetes errors, and recover from them with backoff + +**The problem.** `kube.Backend.Watch` wraps any error from the API server (except the +`sendInitialEvents` refusal) in `fmt.Errorf`. The gateway turns any error that is not a +`*StreamError` into a terminal `INTERNAL`. So `FORBIDDEN`, `UNAUTHENTICATED` and +`UPSTREAM_UNAVAILABLE` never come from the Kubernetes backend. What a browser receives +from `Handler` with a backend that fails as `kube.Backend` does: + +```text +upstream 403: +data: {"seq":1,"type":"error","code":"INTERNAL","message":"krm-stream/kube: streaming list for https://10.43.0.1:443/apis/hello.krm-foyer.example/v1/namespaces/hello/notes: notes.hello.krm-foyer.example is forbidden: User \"oidc:carol@example.com\" cannot watch resource \"notes\"","terminal":true} + +API server unreachable: +data: {"seq":1,"type":"error","code":"INTERNAL","message":"krm-stream/kube: streaming list for https://10.43.0.1:443/api/v1/namespaces/x/configmaps: dial tcp 10.43.0.1:443: connect: connection refused","terminal":true} +``` + +A user without `watch` permission sees "INTERNAL" where Kubernetes said why. A restart of +the API server ends every open stream permanently, though the spec calls that case +retryable. + +**Mapping alone is not enough.** The stream loop answers every non-terminal +`*StreamError` by starting the next cycle at once (`StreamProjection` in +`gateway/stream.go`). It emits a generic `RESYNC_REQUIRED` in place of the error, so the +code and any `retryAfterMs` are lost, and it does not wait. Mapping an unreachable API +server to a non-terminal `UPSTREAM_UNAVAILABLE` would therefore turn an outage into a +tight loop of reconnect attempts. The mapping and the recovery have to change together. + +What we would expect, by the spec's own table (§4.3): + +| Upstream answer | Today | Expected | +| --- | --- | --- | +| 403 Forbidden | `INTERNAL`, terminal | `FORBIDDEN`, terminal, with the API server's `Status.message` | +| 401 Unauthorized (token expired or rejected) | `INTERNAL`, terminal | `UNAUTHENTICATED`, terminal | +| 404 (resource not served, wrong version) | `INTERNAL`, terminal | `SCOPE_INVALID`, terminal | +| Connection refused, timeout, 5xx | `INTERNAL`, terminal | `UPSTREAM_UNAVAILABLE`, not terminal, with backoff | +| 429 from Priority and Fairness | `INTERNAL`, terminal | `UPSTREAM_UNAVAILABLE`, not terminal, `retryAfterMs` from `Retry-After` | +| 410 Gone / `Expired` on an open watch | `RESYNC_REQUIRED` | Unchanged: a new snapshot cycle | +| Any other `ERROR` event on an open watch | `RESYNC_REQUIRED` | Classified as in the rows above; 401 and 403 terminal | + +The mapping is needed on every path an error can take: opening a streaming list, the +LIST and the WATCH of the list-then-watch fallback, and `ERROR` events on an established +watch (`watchError` in `gateway/kube/backend.go` turns all of these into +`RESYNC_REQUIRED` today). + +**Who retries.** The spec leaves open whether a retryable error is retried by the gateway, +on the same connection, or by the client, on a new one. We would prefer the client: the +gateway sends the non-terminal error, with its code and `retryAfterMs`, then closes the +connection. Our reasons: + +- **One retry policy, not two.** The managed connector already has a bounded budget, + exponential backoff and jitter. A second backoff in the gateway would multiply with + it. +- **Nothing is held during an outage.** A gateway that retries keeps the browser's + connection, a goroutine and the host's subscription slot for an outage of unknown + length. A closed connection costs nothing until the client returns. +- **No continuity is lost.** A retry in the gateway starts a new snapshot cycle, and a + reconnect asks for a fresh snapshot anyway, so the browser sees the same thing. + +If you prefer that the gateway retries, it needs a cancellable backoff that honours the +upstream's `Retry-After`, and it must keep the error's code and hint on the wire rather +than replacing them with `RESYNC_REQUIRED`. Either way, please say which in the spec. + +A shared backend's own upstream watch is a different matter: it has no browser to hand +the retry to, so it must retry with backoff itself. + +**Tests we would find convincing:** + +- An API server that keeps refusing connections: the number of upstream attempts per + minute is bounded, and the browser sees `UPSTREAM_UNAVAILABLE`, never `INTERNAL`. +- Cancellation during a backoff, or a closed connection during one, ends it at once. +- 401 and 403, at opening and as watch `ERROR` events, are terminal with the right code. +- 410 on an open watch still starts a new cycle. + +## Ask 2: keep unexpected error text off the wire + +`asStreamError` uses `err.Error()` as the message of any `INTERNAL` it generates, so +whatever an error says reaches the browser. In the output above that is the cluster's +internal address (`10.43.0.1:443`), and, with `NewBackendForConfig`, the configured host, +which `Backend.upstream` says is "never sent to a browser". `Handler` already declines to +pass on a `Principal` error for the same reason. + +We suggest: + +- **A fixed message for a generated `INTERNAL`,** such as "internal error". +- **Messages the host chose stay.** A `*StreamError` returned by an `Authorizer` or + `ClientFor` keeps its message, as does the API server's `Status.message` on a 403 + under ask 1. That message is Kubernetes telling the user about their own access, and + `/k8s` passes it on verbatim too. It is not the same problem as an internal error. +- **A separate diagnostic hook for the raw error,** rather than `Observer`, whose kinds + deliberately carry no error messages (`gateway/observe.go`). The host decides what to + log and what to redact. krm-foyer, for one, logs no token or session ID anywhere, and + "log the full error" would be too broad a default for it. + +A test: an error carrying a recognisable secret string, returned from a backend, appears +in the diagnostic hook and nowhere in the response. + +## Ask 3: let `Handler` serve a host that delegates resource admission + +`ScopePolicy` answers "may anyone stream this kind of thing here?", separately from +the `Authorizer`'s "may this caller see it". That is a legitimate host policy, and the +right default. krm-foyer has a different, equally deliberate one: what its endpoint +exposes is whatever Kubernetes lets the user read, exactly as `/k8s` does. It +promises its users no allowlist. + +`ServeStream` already serves that policy: krm-foyer parses the scope with +`ScopeFromQuery`, checks the target and calls the gateway. So this is not a blocker. We +ask for it because `Handler`'s own comment is right: every host that writes this glue by +hand gets four chances to get security-relevant code subtly wrong, and leaving +`ScopePolicy` out should not mean leaving `Handler` out. + +**A possible shape.** An explicit opt-in that stands out in review, as `AllowAll` does: + +```go +Scopes: gateway.ScopePolicy{ + Targets: []string{""}, + Resources: gateway.AnyResource{}, // admission is left to the upstream's own authorization +}, +``` + +With it, any scope that `ScopeFromQuery` accepts passes `Validate` if its target is +allowlisted. An omitted namespace means whatever the upstream makes of it: the list of a +cluster-scoped resource, or every namespace for a namespaced one, which RBAC allows only +to users with a cluster-wide grant. The `Authorizer` still runs, so a host can narrow it +there. The target allowlist and the refusal of `server`, `url`, `token` and the like stay +mandatory. + +**What it must not promise.** The option should not claim to prove that the stream uses +the caller's credential. A runtime check for `*SharedBackend` would not prove it: a +`kube.Backend` can carry a service account's client, and a wrapper can hide a shared +backend. Binding the stream to the caller's own credential stays the host's obligation, +and the docs should say so where the option is described. + +**The spec.** §8 says the server maps the request to "an allowlisted target + GVR", and +conformance item 10 says the scope is allowlisted. "An allowlisted target and a +host-approved GVR" would cover hosts that approve by delegating to the upstream's own +authorization. + +## Ask 4: let `Principal` say why it failed + +`Handler` answers every `Principal` error with `FORBIDDEN` "not authenticated". A page +treats the two codes differently: `UNAUTHENTICATED` offers sign-in, `FORBIDDEN` says no. +But not every failure means signing in again will help: when a host's session store is +down, the honest answer is "try later". + +We suggest honouring a `*StreamError` that `Principal` returns, as the gateway already +does for `Authorizer` and `ClientFor`, so a host can choose `UNAUTHENTICATED`, +`UPSTREAM_UNAVAILABLE` or `FORBIDDEN` with a message of its own. Any other error keeps +today's generic answer, with the code changed to `UNAUTHENTICATED`, which is what +"not authenticated" says. + +## Ask 5: honour retry hints at the layer that retries + +The managed connector ignores both hints a host can give: `Retry-After` on an HTTP 429, +and `retryAfterMs` on a non-terminal error event. It uses its own backoff for both. + +- **On reconnecting,** the connector should wait at least as long as the last hint + said, within its own cap. krm-foyer's rate bound sends `Retry-After` with the exact + time until a request would be let through, so retries before then are sure to fail. +- **A non-terminal event on an open connection** should not make the client reconnect: + under the gateway's current behaviour, recovery follows on the same connection. The hint + applies only if the connection then ends. +- **If the gateway retries itself** (ask 1), it should honour the upstream's + `Retry-After` in the same way. + +Smaller, and related: on a non-200 answer the connector reports only `stream: HTTP 403` +and discards the body. When the body is a Kubernetes `Status`, as krm-foyer's refusals +are, its `message` would be more useful to show. A bounded read of a JSON `Status`, or a +decoder the host supplies, with today's text as the fallback, would cover it. diff --git a/docs/operations.md b/docs/operations.md index f26bcbf..aa8e530 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -12,6 +12,10 @@ metrics := gateway.ObserverFunc(func(o gateway.Observation) { Do not block in `Observe`; it runs on the stream or shared-watch goroutine. Never label a metric with object name, UID, principal, patch contents, or an error message. +Error text has its own hook, `Options.Diagnostics`, because it can name internal addresses and URLs. +The browser receives only the protocol code and a message chosen for it; an unexpected error is sent +as `INTERNAL` "internal error". Decide what to log there, and redact it. + ## Signals to alert on | Signal | Meaning | First response | @@ -20,6 +24,7 @@ object name, UID, principal, patch contents, or an error message. | `consumer_resync` rising | upstream continuity was lost or a shared subscriber fell behind | correlate with API-server errors, shared overflows, and deployments | | `shared_overflow` | one subscriber exceeded `SharedOptions.QueueDepth` | increase only after checking browser stalls and event rate; resnapshot is intentional | | `terminal_error` | logical-stream failure, observed before attempting its terminal frame; delivery can fail | alert by low-cardinality error code; browsers must not retry terminal errors | +| `retryable_error` rising | streams ending with a non-terminal error, usually `UPSTREAM_UNAVAILABLE` | check API-server health and priority-and-fairness rejections; browsers reconnect on their own backoff | | `event_suppressed` ratio | `krm-spec/v1` is removing expected churn | a sharp drop may mean callers selected `krm-full/v1` or a projection changed | | stream count / snapshot duration | connection pressure or oversized scopes | narrow namespaces/selectors; avoid accidental all-namespaces watches | | unorderable `resourceVersion` terminal errors | an unsupported or aggregated API does not meet strict ordering | use `OrderingLenient` only after accepting the reduced monotonicity guarantee | @@ -34,6 +39,7 @@ object name, UID, principal, patch contents, or an error message. | `ScopePolicy.AllowLabelSelector` | false | enable only for an endpoint that deliberately supports caller narrowing | | `GroupResource.AllowAllNamespaces` | false | make all-namespaces access an explicit reviewable policy decision | | `Gateway.Ordering` | strict | keep strict on supported Kubernetes; use lenient only for known aggregated APIs | +| `gateway.Options.Diagnostics` | nil (discarded) | receive the raw error behind each error event; redact before logging | Snapshot object and byte limits remain a host-level scope policy concern. The gateway refuses to guess a safe universal cap: object size, useful namespace size, and recovery behavior are product-specific. diff --git a/docs/releasing.md b/docs/releasing.md index 6635671..573465a 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -24,6 +24,19 @@ produces these tags and packages: Only `@configbutler/krm-stream` is maintained on npm. Do not publish to the unscoped `krm-stream` name. +`gateway/kube/go.mod` requires the core module by version, and `go.work` hides that version in this +checkout. When an adapter change needs core API or behaviour that is not released yet, push the core +change first, then point the adapter at that commit from outside the workspace: + +```sh +cd gateway/kube +GOWORK=off GOPROXY=direct go get github.com/ConfigButler/krm-stream/gateway@ +GOWORK=off go mod tidy +``` + +Check it from a clean module outside this workspace, as CI's `a stranger can go get this` job does. +After the next release, move the requirement to the released core tag. + Before 1.0, remove superseded API names and forwarding packages instead of maintaining compatibility shims. Record each removal and its replacement in release notes, and update repository callers, examples and documentation together. Changes to wire semantics still require explicit spec and diff --git a/gateway/CHANGELOG.md b/gateway/CHANGELOG.md index 68774ad..eed2f64 100644 --- a/gateway/CHANGELOG.md +++ b/gateway/CHANGELOG.md @@ -13,6 +13,21 @@ own headers, delivery and cleanup. `SSESink.Heartbeat` now returns an error; callers must handle failure and stop their stream. `NewSSESink(io.Writer)` remains generic and propagates available flush errors without installing HTTP deadlines. The v1 wire protocol is unchanged. +- Recover only `RESYNC_REQUIRED` on the open connection. Any other non-terminal error, such as + `UPSTREAM_UNAVAILABLE`, is sent with its code and `retryAfterMs` and then closes the connection, + so the client owns the retry. A new `retryable_error` observation reports it. +- `SharedBackend` backs off a scope whose upstream fails with `UPSTREAM_UNAVAILABLE`, whether opening + fails or the open watch dies of it: one attempt per period (1s doubling to 30s, or the upstream's + hint) however many subscribers reconnect. A completed upstream snapshot resets it. +- Add `Options.Diagnostics` / `Gateway.Diagnostics`, which receive the raw error behind each error + event, and `StreamError.Cause`, which never reaches the wire. Add `Unauthenticated` and + `UpstreamUnavailable` constructors. +- Add `ScopePolicy.AnyResource`, which leaves resource admission to the upstream's authorization. + Setting it with `Resources` panics in `Handler` and refuses every scope in `Validate`. +- `Principal` may return a `*StreamError` to choose its refusal. +- **Breaking:** a generated `INTERNAL` says "internal error" rather than the underlying error's + text. Any other `Principal` error is `UNAUTHENTICATED` rather than `FORBIDDEN`. A timed + reauthorization that fails with `UPSTREAM_UNAVAILABLE` ends the stream without making it terminal. ## [0.4.0](https://github.com/ConfigButler/krm-stream/compare/gateway/v0.3.0...gateway/v0.4.0) (2026-09-11) diff --git a/gateway/README.md b/gateway/README.md index 7afc1b4..d24189d 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -15,7 +15,7 @@ right upstream for the host application. | `Principal` | Resolve the HTTP request to an application principal. | | `Authorizer` | Allow or deny the normalized scope before a watch opens. | | `Clients` | Return a backend acting as the caller, or an explicitly shared backend. | -| `Scopes` | Allowlist target and Kubernetes group/resource combinations. | +| `Scopes` | Allowlist targets, and allowlist group/resource combinations or delegate them to Kubernetes. | The zero `ScopePolicy` denies every request. The gateway never accepts an API-server URL or a credential from a browser request. @@ -56,7 +56,17 @@ narrowing only; a host should still constrain selector complexity at its HTTP bo Every snapshot cycle emits `reset`, zero or more `added` events, then `synced`. Live updates are complete-object replacements. A recoverable upstream discontinuity emits `RESYNC_REQUIRED` and starts -a new cycle on the same SSE connection. Terminal errors are the final event and close the connection. +a new cycle on the same SSE connection. Any other non-terminal error, such as `UPSTREAM_UNAVAILABLE`, +is sent with its `retryAfterMs` and then closes the connection: the client owns that retry. Terminal +errors are the final event and close the connection. + +An unexpected error reaches the browser as `INTERNAL` with the message `internal error`. Its text, +and the `Cause` of any `StreamError`, go only to `Options.Diagnostics`, so the host decides what to +log and what to redact. `gateway/kube` maps API-server failures to protocol codes: 403 to +`FORBIDDEN` with Kubernetes' own message, 401 to `UNAUTHENTICATED`, 404 to `SCOPE_INVALID`, and 429, +5xx or an unreachable server to `UPSTREAM_UNAVAILABLE` (with `Retry-After` as `retryAfterMs`). +`SharedBackend` backs off a scope whose upstream will not open, so its subscribers' reconnects do +not add up to a stampede. The gateway absorbs Kubernetes-specific mechanics including bookmarks, relists, 410 responses, partial metadata objects, and ambiguous deletion tombstones. The normative details are in diff --git a/gateway/event.go b/gateway/event.go index de64c0c..c002ff8 100644 --- a/gateway/event.go +++ b/gateway/event.go @@ -49,7 +49,7 @@ const ( CodeForbidden ErrorCode = "FORBIDDEN" // terminal: lost (or never had) access CodeUnauthenticated ErrorCode = "UNAUTHENTICATED" // terminal: credential expired/rejected CodeScopeInvalid ErrorCode = "SCOPE_INVALID" // terminal: scope not allowlisted - CodeUpstreamUnavailable ErrorCode = "UPSTREAM_UNAVAILABLE" // retryable: can't reach the API server + CodeUpstreamUnavailable ErrorCode = "UPSTREAM_UNAVAILABLE" // retryable by the client: the gateway closes after it CodeResyncRequired ErrorCode = "RESYNC_REQUIRED" // retryable: continuity lost, a new cycle follows CodeSlowConsumer ErrorCode = "SLOW_CONSUMER" // terminal: fell too far behind, dropped CodeInternal ErrorCode = "INTERNAL" // either; `terminal` says whether to give up @@ -101,8 +101,8 @@ type Identity struct { } // Scope is the normalized, server-validated target of a stream. The browser asks for a LOGICAL -// scope; the server maps it to an allowlisted target + GVR + a namespace the caller may actually -// see. A raw API-server URL is never accepted. +// scope; the server maps it to an allowlisted target, a host-approved GVR and a namespace the caller +// may actually see. A raw API-server URL is never accepted. type Scope struct { Target string `json:"target"` Group string `json:"group,omitempty"` diff --git a/gateway/handler.go b/gateway/handler.go index af9434b..cec7e8a 100644 --- a/gateway/handler.go +++ b/gateway/handler.go @@ -2,6 +2,7 @@ package gateway import ( "context" + "errors" "net/http" "time" ) @@ -35,6 +36,12 @@ type Options struct { // Principal answers "who is calling?" from the request — a session cookie, an mTLS peer, a // header your ingress set. Return an error to refuse. // + // Return a *StreamError to choose what the caller is told: Unauthenticated when signing in again + // will help, UpstreamUnavailable when your session store is down and the honest answer is "try + // later". Any other error is sent as UNAUTHENTICATED "not authenticated", and its text goes only + // to Diagnostics: what went wrong identifying a caller is your business, and possibly your + // internals. + // // Required, and there is deliberately no default. Every default that could go here ("nil // principal", "the request's Authorization header") is a policy decision about someone's auth // system, and this library does not get to make one. @@ -48,7 +55,7 @@ type Options struct { Clients ClientFor // Scopes is the allowlist (spec §8). Required, and deny-by-default: the zero value streams - // nothing, so forgetting it fails closed. + // nothing, so forgetting it fails closed. Setting both Resources and AnyResource panics. Scopes ScopePolicy // Projection defaults to ProjectionFull — the safe one. A gateway that defaulted to raw and @@ -65,6 +72,10 @@ type Options struct { // Observer receives low-cardinality stream lifecycle signals. It must not block. Observer Observer + // Diagnostics receives the raw error behind every error event, including the detail the wire + // does not carry. It must not block. Nil discards it. + Diagnostics Diagnostics + // HeartbeatInterval defaults to HeartbeatInterval. Set a positive value to match a proxy's idle // timeout; it affects HTTP streams only. HeartbeatInterval time.Duration @@ -100,6 +111,8 @@ func Handler(o Options) http.Handler { panic("krm-stream: Options.Authorizer is required — use gateway.AllowAll{} to say you meant it") case o.Clients == nil: panic("krm-stream: Options.Clients is required — the library holds no cluster connection of its own") + case o.Scopes.AnyResource && len(o.Scopes.Resources) > 0: + panic("krm-stream: ScopePolicy sets both Resources and AnyResource — choose an allowlist or delegation, not both") } g := &Gateway{ @@ -109,6 +122,7 @@ func Handler(o Options) http.Handler { Projections: o.Projections, Ordering: o.Ordering, Observer: o.Observer, + Diagnostics: o.Diagnostics, HeartbeatInterval: o.HeartbeatInterval, WriteTimeout: o.WriteTimeout, ReauthorizationInterval: o.ReauthorizationInterval, @@ -118,10 +132,11 @@ func Handler(o Options) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { principal, err := o.Principal(r) if err != nil { - // Forbidden, not the host's error verbatim: whatever went wrong identifying this caller - // is the host's business and possibly its internals. The caller learns that they may not - // have this, and nothing else. - g.refuse(w, r, Forbidden("not authenticated")) + var se *StreamError + if !errors.As(err, &se) || se == nil { + se = &StreamError{Code: CodeUnauthenticated, Message: "not authenticated", Terminal: true, Cause: err} + } + g.refuse(w, r, Diagnostic{Code: se.Code, Terminal: se.Terminal, Err: err}, se) return } @@ -130,7 +145,8 @@ func Handler(o Options) http.Handler { err = o.Scopes.Validate(scope) } if err != nil { - g.refuse(w, r, asStreamError(err)) + se := asStreamError(err) + g.refuse(w, r, Diagnostic{Principal: principal, Scope: scope, Code: se.Code, Terminal: se.Terminal, Err: err}, se) return } @@ -138,8 +154,9 @@ func Handler(o Options) http.Handler { }) } -// refuse writes a terminal error as a well-formed one-event stream, and closes. -func (g *Gateway) refuse(w http.ResponseWriter, r *http.Request, serr *StreamError) { +// refuse writes an error as a well-formed one-event stream, and closes. +func (g *Gateway) refuse(w http.ResponseWriter, r *http.Request, d Diagnostic, serr *StreamError) { + g.diagnose(d) g.serveHTTP(w, r, func(ctx context.Context, sink *SSESink) { _ = (&sequenceSink{sink: sink}).Emit(ctx, serr.Event()) }) diff --git a/gateway/handler_test.go b/gateway/handler_test.go index c0795df..cac7513 100644 --- a/gateway/handler_test.go +++ b/gateway/handler_test.go @@ -2,6 +2,7 @@ package gateway_test import ( "context" + "errors" "net/http" "net/http/httptest" "strings" @@ -117,7 +118,7 @@ func TestHandlerRefusalsAreTerminalStreamEvents(t *testing.T) { { name: "a caller we cannot identify", target: "/s?version=v1&resource=configmaps&namespace=app", - code: "FORBIDDEN", + code: "UNAUTHENTICATED", opts: func() gateway.Options { o := testOptions(configmapsAllowed) o.Principal = func(*http.Request) (gateway.Principal, error) { @@ -152,7 +153,7 @@ func TestHandlerAuthenticatesBeforeDescribingScopeFailures(t *testing.T) { o := testOptions(configmapsAllowed) o.Principal = func(*http.Request) (gateway.Principal, error) { return nil, http.ErrNoCookie } rec := serve(t, o, "/s?resource=secrets") // malformed and disallowed, but caller is unauthenticated - if body := rec.Body.String(); !strings.Contains(body, `"code":"FORBIDDEN"`) || strings.Contains(body, `"code":"SCOPE_INVALID"`) { + if body := rec.Body.String(); !strings.Contains(body, `"code":"UNAUTHENTICATED"`) || strings.Contains(body, `"code":"SCOPE_INVALID"`) { t.Errorf("unauthenticated request disclosed scope validation: %s", body) } } @@ -190,3 +191,117 @@ func TestHandlerPanicsOnMissingSeams(t *testing.T) { }) } } + +// A host chooses what an unidentified caller is told. "Sign in again" is wrong when the host's session +// store is down; "try later" is the honest answer, and it is not terminal. +func TestPrincipalChoosesItsRefusal(t *testing.T) { + o := testOptions(configmapsAllowed) + o.Principal = func(*http.Request) (gateway.Principal, error) { + return nil, gateway.UpstreamUnavailable("session store unavailable", 2*time.Second) + } + body := serve(t, o, "/s?version=v1&resource=configmaps&namespace=app").Body.String() + for _, want := range []string{`"code":"UPSTREAM_UNAVAILABLE"`, `"terminal":false`, `"retryAfterMs":2000`, `session store unavailable`} { + if !strings.Contains(body, want) { + t.Errorf("want %s in:\n%s", want, body) + } + } +} + +func TestPrincipalErrorTextReachesOnlyDiagnostics(t *testing.T) { + const secret = "redis://session-store.internal:6379" + var got []gateway.Diagnostic + o := testOptions(configmapsAllowed) + o.Principal = func(*http.Request) (gateway.Principal, error) { return nil, errors.New("dial " + secret) } + o.Diagnostics = func(d gateway.Diagnostic) { got = append(got, d) } + + body := serve(t, o, "/s?version=v1&resource=configmaps&namespace=app").Body.String() + if strings.Contains(body, secret) { + t.Fatalf("the host's error text reached the wire:\n%s", body) + } + if len(got) != 1 || got[0].Code != gateway.CodeUnauthenticated || !strings.Contains(got[0].Err.Error(), secret) { + t.Fatalf("diagnostics = %+v, want one UNAUTHENTICATED carrying the raw error", got) + } +} + +// An unexpected backend error is INTERNAL, and its text may name anything: the cluster's internal +// address, a URL a backend promised never to show a browser. The wire says "internal error"; the host +// sees the rest. +func TestUnexpectedErrorTextReachesOnlyDiagnostics(t *testing.T) { + const internalURL = "https://10.43.0.1:443/api/v1/namespaces/app/configmaps" + var got []gateway.Diagnostic + o := testOptions(configmapsAllowed) + o.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend, error) { + return failingBackend{errors.New("streaming list for " + internalURL + ": boom")}, nil + } + o.Diagnostics = func(d gateway.Diagnostic) { got = append(got, d) } + + body := serve(t, o, "/s?version=v1&resource=configmaps&namespace=app").Body.String() + if strings.Contains(body, internalURL) || !strings.Contains(body, `"message":"internal error"`) { + t.Fatalf("want a generic INTERNAL, got:\n%s", body) + } + if len(got) != 1 || got[0].Principal != "alice" || got[0].Scope.Resource != "configmaps" || !strings.Contains(got[0].Err.Error(), internalURL) { + t.Fatalf("diagnostics = %+v, want the raw error with its principal and scope", got) + } +} + +type failingBackend struct{ err error } + +func (b failingBackend) Watch(context.Context, gateway.Scope) (gateway.Watcher, error) { + return nil, b.err +} + +// A retryable upstream error is sent with its code and hint, and then the connection closes: the +// client owns the retry. The handler must not resnapshot in a loop against a dead API server. +func TestRetryableUpstreamErrorEndsTheConnection(t *testing.T) { + attempts := 0 + o := testOptions(configmapsAllowed) + o.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend, error) { + attempts++ + return failingBackend{gateway.UpstreamUnavailable("the API server is unavailable", 3*time.Second)}, nil + } + body := serve(t, o, "/s?version=v1&resource=configmaps&namespace=app").Body.String() + if attempts != 1 { + t.Errorf("upstream attempts = %d, want 1: the gateway retried on its own", attempts) + } + want := `data: {"seq":1,"type":"error","code":"UPSTREAM_UNAVAILABLE","message":"the API server is unavailable","retryAfterMs":3000,"terminal":false}` + if strings.TrimSpace(body) != want { + t.Errorf("body:\n%s\nwant:\n%s", body, want) + } +} + +// AnyResource is a host policy that delegates resource admission to the upstream's own +// authorization. It admits any resource, with or without a namespace, but keeps the target allowlist. +func TestAnyResourceDelegatesResourceAdmission(t *testing.T) { + policy := gateway.ScopePolicy{Targets: []string{""}, AnyResource: true} + for _, target := range []string{ + "/s?group=hello.krm-foyer.example&version=v1&resource=notes&namespace=hello", + "/s?version=v1&resource=namespaces", + "/s?version=v1&resource=configmaps", + } { + if body := serve(t, testOptions(policy), target).Body.String(); !strings.Contains(body, `"type":"synced"`) { + t.Errorf("%s was not streamed:\n%s", target, body) + } + } + for _, target := range []string{ + "/s?target=other&version=v1&resource=configmaps&namespace=app", + "/s?version=v1&resource=configmaps&namespace=app&labelSelector=a%3Db", + "/s?version=v1&resource=configmaps&server=https://10.0.0.1:6443", + } { + if body := serve(t, testOptions(policy), target).Body.String(); !strings.Contains(body, `"code":"SCOPE_INVALID"`) { + t.Errorf("%s was not refused:\n%s", target, body) + } + } +} + +func TestAnyResourceWithAnAllowlistIsRefused(t *testing.T) { + policy := gateway.ScopePolicy{Targets: []string{""}, AnyResource: true, Resources: configmapsAllowed.Resources} + if err := policy.Validate(gateway.Scope{Version: "v1", Resource: "configmaps", Namespace: "app"}); err == nil { + t.Error("Validate admitted a scope under an ambiguous policy") + } + defer func() { + if recover() == nil { + t.Error("Handler mounted an ambiguous scope policy") + } + }() + gateway.Handler(testOptions(policy)) +} diff --git a/gateway/kube/CHANGELOG.md b/gateway/kube/CHANGELOG.md index 367965e..efa8ccb 100644 --- a/gateway/kube/CHANGELOG.md +++ b/gateway/kube/CHANGELOG.md @@ -6,6 +6,15 @@ session expiry, service-account SARs, bounded SSE delivery and lifecycle counters. It requires a verified-HTTPS cluster configuration, since it carries service and participant tokens. Includes a manually invoked independent-identity Kubernetes fixture; no new public identity helper. +- Map API-server failures to protocol codes on every path (streaming list, the fallback's LIST and + WATCH, watch `ERROR` events): 403 `FORBIDDEN` with Kubernetes' message, 401 `UNAUTHENTICATED`, + 404 `SCOPE_INVALID`, 429, 5xx and unreachable servers `UPSTREAM_UNAVAILABLE` with `Retry-After` as + `retryAfterMs`. A 410 still starts a new snapshot cycle. The raw error, with the URL that failed, + is the `Cause`. +- A SubjectAccessReview that cannot reach the API server, or is throttled, is `UPSTREAM_UNAVAILABLE` + and keeps its retry hint. +- **Breaking:** `SubjectAccessReviewAuthorizer` refuses a principal it cannot map with + `UNAUTHENTICATED` rather than `FORBIDDEN`. ## [0.4.0](https://github.com/ConfigButler/krm-stream/compare/gateway/kube/v0.3.0...gateway/kube/v0.4.0) (2026-09-11) diff --git a/gateway/kube/authz.go b/gateway/kube/authz.go index 0a61660..d84d238 100644 --- a/gateway/kube/authz.go +++ b/gateway/kube/authz.go @@ -66,7 +66,7 @@ func SubjectAccessReviewAuthorizer(cs kubernetes.Interface, subjectFor SubjectFo return gateway.AuthorizerFunc(func(ctx context.Context, p gateway.Principal, scope gateway.Scope) error { subject, err := subjectFor(p) if err != nil { - return gateway.Forbidden("not authenticated") + return gateway.Unauthenticated("not authenticated") } if subject.User == "" && len(subject.Groups) == 0 { // An empty subject is not "anonymous, and therefore probably fine" — it is a request that @@ -79,7 +79,15 @@ func SubjectAccessReviewAuthorizer(cs kubernetes.Interface, subjectFor SubjectFo if err != nil { // A SubjectAccessReview we could not complete is NOT an allow. If the API server // cannot tell us whether this caller may look, the answer is no. - return fmt.Errorf("krm-stream/kube: subject access review (%s): %w", verb, err) + // + // An unreachable API server is UPSTREAM_UNAVAILABLE: the stream ends, and the client + // may come back. Anything else (a gateway identity that may not create reviews, say) + // is the host's misconfiguration, and stays a generic INTERNAL. + err = fmt.Errorf("krm-stream/kube: subject access review (%s): %w", verb, err) + if se, ok := classify(err).(*gateway.StreamError); ok && se.Code == gateway.CodeUpstreamUnavailable { + return se // with its Retry-After hint and cause intact + } + return err } if !allowed { // The API server's own words, so an operator can find this decision in the audit log diff --git a/gateway/kube/authz_test.go b/gateway/kube/authz_test.go index fbd671b..0c0ec92 100644 --- a/gateway/kube/authz_test.go +++ b/gateway/kube/authz_test.go @@ -3,9 +3,13 @@ package kube_test import ( "context" "errors" + "net/http" + "net/http/httptest" + "strings" "testing" authzv1 "k8s.io/api/authorization/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/kubernetes/fake" k8stesting "k8s.io/client-go/testing" @@ -170,3 +174,31 @@ func TestANamedScopeAsksAboutThatName(t *testing.T) { } } } + +// A review the API server throttles is retryable, and keeps the wait the API server asked for, all +// the way to the event the browser reads. +func TestAThrottledReviewKeepsItsRetryHint(t *testing.T) { + cs := fake.NewSimpleClientset() + cs.PrependReactor("create", "subjectaccessreviews", + func(k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, apierrors.NewTooManyRequests("priority and fairness", 7) + }) + handler := gateway.Handler(gateway.Options{ + Principal: func(*http.Request) (gateway.Principal, error) { return alice, nil }, + Authorizer: kube.SubjectAccessReviewAuthorizer(cs, subjectOf), + Clients: func(context.Context, string, gateway.Principal) (gateway.Backend, error) { + t.Fatal("a throttled review opened a watch") + return nil, nil + }, + Scopes: gateway.ScopePolicy{Targets: []string{""}, AnyResource: true}, + }) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/s?version=v1&resource=configmaps&namespace=app", nil)) + + body := rec.Body.String() + for _, want := range []string{`"code":"UPSTREAM_UNAVAILABLE"`, `"retryAfterMs":7000`, `"terminal":false`} { + if !strings.Contains(body, want) { + t.Errorf("want %s in:\n%s", want, body) + } + } +} diff --git a/gateway/kube/backend.go b/gateway/kube/backend.go index 792002b..4830128 100644 --- a/gateway/kube/backend.go +++ b/gateway/kube/backend.go @@ -25,16 +25,21 @@ package kube import ( "context" + "crypto/tls" + "crypto/x509" "errors" "fmt" + "net" "strings" "sync" + "time" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" + utilnet "k8s.io/apimachinery/pkg/util/net" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/dynamic" "k8s.io/client-go/rest" @@ -58,7 +63,8 @@ type Backend struct { client dynamic.Interface // host is what this backend dialed, for error messages ONLY — it is never sent to a browser and - // never parsed. An upstream error that names neither the server nor the resource path reads like + // never parsed. Errors carry it as a StreamError's Cause, which reaches only the host's + // Diagnostics hook. An upstream error that names neither the server nor the resource path reads like // "that resource does not exist", when what actually happened may be "you built the wrong URL". // A kcp workspace, whose kubeconfig carries a path prefix, is exactly where the two are // indistinguishable without this. @@ -146,7 +152,7 @@ func (b *Backend) Watch(ctx context.Context, scope gateway.Scope) (gateway.Watch // is NOT ours to paper over. Falling back on every failure would turn "you may not watch // Secrets" into a second, differently-worded permission denial, which is how a gateway ends // up lying about why it could not open a stream. - return nil, fmt.Errorf("krm-stream/kube: streaming list for %s: %w", b.upstream(scope), err) + return nil, classify(fmt.Errorf("krm-stream/kube: streaming list for %s: %w", b.upstream(scope), err)) } // F6, in production. This API is aggregated (or otherwise has WatchList off); list-then-watch is not a @@ -212,7 +218,7 @@ func (b *Backend) rememberListThenWatch(gv schema.GroupVersion) { func (b *Backend) listThenWatchStream(ctx context.Context, ri dynamic.ResourceInterface, scope gateway.Scope) (gateway.Watcher, error) { list, err := ri.List(ctx, selectors(scope)) if err != nil { - return nil, fmt.Errorf("krm-stream/kube: list %s: %w", b.upstream(scope), err) + return nil, classify(fmt.Errorf("krm-stream/kube: list %s: %w", b.upstream(scope), err)) } o := selectors(scope) @@ -220,8 +226,8 @@ func (b *Backend) listThenWatchStream(ctx context.Context, ri dynamic.ResourceIn o.ResourceVersion = list.GetResourceVersion() // resume EXACTLY where the list ended: no gap. w, err := ri.Watch(ctx, o) if err != nil { - return nil, fmt.Errorf("krm-stream/kube: watch %s from resourceVersion %q: %w", - b.upstream(scope), list.GetResourceVersion(), err) + return nil, classify(fmt.Errorf("krm-stream/kube: watch %s from resourceVersion %q: %w", + b.upstream(scope), list.GetResourceVersion(), err)) } // The snapshot, in the shape a streaming list would deliver — including the boundary bookmark, @@ -364,17 +370,88 @@ func translate(ev watch.Event) (gateway.WatchEvent, error) { } } -// watchError renders the *metav1.Status on a watch.Error as an error the gateway can act on. A 410 -// Gone / "Expired" is the continuity-losing one, and it is not fatal: it means "start a new snapshot -// cycle", which is exactly what a non-terminal RESYNC_REQUIRED tells the consumer. +// watchError renders the *metav1.Status on a watch.Error as an error the gateway can act on, by the +// same rules as a failure to open the watch: a revoked grant on an open watch is as terminal as one +// at opening. A 410 Gone / "Expired" is the continuity-losing one, and it is not fatal: it means +// "start a new snapshot cycle", which is exactly what a non-terminal RESYNC_REQUIRED tells the +// consumer. So is anything this adapter cannot classify. func watchError(obj runtime.Object) error { - if status, ok := obj.(*metav1.Status); ok { - if status.Reason == metav1.StatusReasonExpired || status.Code == 410 { - return gateway.ResyncRequired(fmt.Sprintf( - "the upstream resourceVersion expired (410 Gone): %s", status.Message)) + status, ok := obj.(*metav1.Status) + if !ok { + return gateway.ResyncRequired(fmt.Sprintf("upstream watch error carrying %T", obj)) + } + cause := fmt.Errorf("krm-stream/kube: watch error event: %w", apierrors.FromObject(status)) + if se := classifyStatus(cause); se != nil { + return se + } + resync := gateway.ResyncRequired("upstream watch error; a new snapshot cycle follows") + resync.Cause = cause + return resync +} + +// classify maps an API-server failure onto the spec's error codes (§4.3). What it cannot classify it +// returns unchanged, which the gateway sends as a generic INTERNAL; the full error, with the URL that +// failed, always survives as the Cause, for the host's Diagnostics and never for the wire. +func classify(err error) error { + if se := classifyStatus(err); se != nil { + return se + } + if unavailable(err) { + return withCause(gateway.UpstreamUnavailable("the API server is unavailable", 0), err) + } + return err +} + +// classifyStatus maps an API-server Status, or nil when err carries none it recognises. +func classifyStatus(err error) *gateway.StreamError { + var status apierrors.APIStatus + if !errors.As(err, &status) { + return nil + } + switch { + case apierrors.IsResourceExpired(err) || apierrors.IsGone(err): + return withCause(gateway.ResyncRequired("the upstream resourceVersion expired (410 Gone)"), err) + case apierrors.IsUnauthorized(err): + return withCause(gateway.Unauthenticated("the API server rejected the caller's credential"), err) + case apierrors.IsForbidden(err): + // Kubernetes' own answer, verbatim: it tells the caller about their own access, which is what + // a `kubectl` user or a `/k8s` proxy would see too. It names the resource and the subject, not + // the server. + msg := status.Status().Message + if msg == "" { + msg = "the API server refused this watch" + } + return withCause(gateway.Forbidden(msg), err) + case apierrors.IsNotFound(err): + return withCause(gateway.ScopeInvalid("the API server does not serve this resource at this version"), err) + case apierrors.IsTooManyRequests(err): + var retryAfter time.Duration + if seconds, ok := apierrors.SuggestsClientDelay(err); ok { + retryAfter = time.Duration(seconds) * time.Second } - return gateway.ResyncRequired(fmt.Sprintf("upstream watch error: %s (reason=%s code=%d)", - status.Message, status.Reason, status.Code)) + return withCause(gateway.UpstreamUnavailable("the API server is overloaded", retryAfter), err) + case status.Status().Code >= 500 || apierrors.IsServerTimeout(err) || apierrors.IsTimeout(err): + return withCause(gateway.UpstreamUnavailable("the API server is unavailable", 0), err) } - return gateway.ResyncRequired(fmt.Sprintf("upstream watch error carrying %T", obj)) + return nil +} + +// unavailable reports a transport failure: the API server could not be reached, or went away +// mid-request. A certificate the client will not trust is not one of those. It is a configuration +// error that no retry will fix, so it stays an INTERNAL. +func unavailable(err error) bool { + var certInvalid x509.CertificateInvalidError + var unknownAuthority x509.UnknownAuthorityError + var hostname x509.HostnameError + var verification *tls.CertificateVerificationError + if errors.As(err, &certInvalid) || errors.As(err, &unknownAuthority) || errors.As(err, &hostname) || errors.As(err, &verification) { + return false + } + var netErr net.Error + return errors.As(err, &netErr) || utilnet.IsConnectionRefused(err) || utilnet.IsConnectionReset(err) || utilnet.IsProbableEOF(err) +} + +func withCause(se *gateway.StreamError, cause error) *gateway.StreamError { + se.Cause = cause + return se } diff --git a/gateway/kube/classify_test.go b/gateway/kube/classify_test.go new file mode 100644 index 0000000..f1ac31d --- /dev/null +++ b/gateway/kube/classify_test.go @@ -0,0 +1,140 @@ +package kube + +import ( + "errors" + "net" + "strings" + "testing" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/rest" + + "github.com/ConfigButler/krm-stream/gateway" +) + +// Every way an API server can refuse, on every path an error can take, maps to the spec's code +// (§4.3). Before this, every one of them reached a browser as a terminal INTERNAL carrying the +// cluster's internal URL. +func TestAPIServerErrorsMapToProtocolCodes(t *testing.T) { + gr := schema.GroupResource{Group: "hello.krm-foyer.example", Resource: "notes"} + for _, tc := range []struct { + name string + err error + code gateway.ErrorCode + terminal bool + retryAfter int + message string + }{ + {"403", apierrors.NewForbidden(gr, "", errors.New(`User "oidc:carol@example.com" cannot watch resource "notes"`)), + gateway.CodeForbidden, true, 0, `cannot watch resource "notes"`}, + {"401", apierrors.NewUnauthorized("token expired"), gateway.CodeUnauthenticated, true, 0, ""}, + {"404", apierrors.NewNotFound(gr, ""), gateway.CodeScopeInvalid, true, 0, ""}, + {"429", apierrors.NewTooManyRequests("slow down", 7), gateway.CodeUpstreamUnavailable, false, 7000, ""}, + {"500", apierrors.NewInternalError(errors.New("etcd")), gateway.CodeUpstreamUnavailable, false, 0, ""}, + {"503", apierrors.NewServiceUnavailable("restarting"), gateway.CodeUpstreamUnavailable, false, 0, ""}, + {"504", apierrors.NewTimeoutError("slow", 0), gateway.CodeUpstreamUnavailable, false, 0, ""}, + {"connection refused", &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connect: connection refused")}, + gateway.CodeUpstreamUnavailable, false, 0, ""}, + } { + for _, path := range []string{"streaming list", "fallback list", "fallback watch"} { + t.Run(tc.name+"/"+path, func(t *testing.T) { + client, res := newStub(t) + res.watchFn = func(o metav1.ListOptions) (watch.Interface, error) { + if path == "streaming list" || (path == "fallback watch" && o.SendInitialEvents == nil) { + return nil, tc.err + } + return nil, refusal() + } + res.listFn = func(metav1.ListOptions) (*unstructured.UnstructuredList, error) { + if path == "fallback list" { + return nil, tc.err + } + return &unstructured.UnstructuredList{}, nil + } + _, err := NewBackend(client).Watch(t.Context(), scope) + _ = assertCode(t, err, tc.code, tc.terminal, tc.retryAfter, tc.message) + }) + } + } +} + +// The same rules hold for an ERROR event on an established watch: a grant revoked mid-stream is as +// terminal as one refused at opening. Only 410, and what cannot be classified, start a new cycle. +func TestWatchErrorEventsAreClassified(t *testing.T) { + for _, tc := range []struct { + status metav1.Status + code gateway.ErrorCode + terminal bool + }{ + {metav1.Status{Code: 403, Reason: metav1.StatusReasonForbidden, Message: "revoked"}, gateway.CodeForbidden, true}, + {metav1.Status{Code: 401, Reason: metav1.StatusReasonUnauthorized}, gateway.CodeUnauthenticated, true}, + {metav1.Status{Code: 500, Reason: metav1.StatusReasonInternalError}, gateway.CodeUpstreamUnavailable, false}, + {metav1.Status{Code: 410, Reason: metav1.StatusReasonExpired}, gateway.CodeResyncRequired, false}, + {metav1.Status{Code: 422, Reason: metav1.StatusReasonInvalid}, gateway.CodeResyncRequired, false}, + } { + t.Run(string(tc.status.Reason), func(t *testing.T) { + client, res := newStub(t) + fake := watch.NewFakeWithChanSize(1, false) + res.watchFn = func(metav1.ListOptions) (watch.Interface, error) { return fake, nil } + w, err := NewBackend(client).Watch(t.Context(), scope) + if err != nil { + t.Fatalf("Watch: %v", err) + } + defer w.Stop() + status := tc.status + status.Status = metav1.StatusFailure + fake.Error(&status) + _ = assertCode(t, drain(t, w, 1)[0].Err, tc.code, tc.terminal, 0, "") + }) + } +} + +// A real dial against a port nobody listens on: the browser hears UPSTREAM_UNAVAILABLE, and nothing +// about where the gateway was dialing. The address survives only as the Cause. +func TestAnUnreachableAPIServerIsUpstreamUnavailable(t *testing.T) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + addr := l.Addr().String() + _ = l.Close() + + b, err := NewBackendForConfig(&rest.Config{Host: "http://" + addr}) + if err != nil { + t.Fatal(err) + } + _, err = b.Watch(t.Context(), scope) + se := assertCode(t, err, gateway.CodeUpstreamUnavailable, false, 0, "") + if strings.Contains(se.Message, addr) || !strings.Contains(se.Cause.Error(), addr) { + t.Errorf("message %q must not name %s; the cause %q must", se.Message, addr, se.Cause) + } +} + +func assertCode(t *testing.T, err error, code gateway.ErrorCode, terminal bool, retryAfter int, message string) *gateway.StreamError { + t.Helper() + var se *gateway.StreamError + if !errors.As(err, &se) { + t.Fatalf("err = %v (%T), want a *gateway.StreamError", err, err) + } + if se.Code != code || se.Terminal != terminal { + t.Errorf("got %s terminal=%v, want %s terminal=%v (%v)", se.Code, se.Terminal, code, terminal, err) + } + got := 0 + if se.RetryAfterMs != nil { + got = *se.RetryAfterMs + } + if got != retryAfter { + t.Errorf("retryAfterMs = %d, want %d", got, retryAfter) + } + if !strings.Contains(se.Message, message) { + t.Errorf("message = %q, want it to contain %q", se.Message, message) + } + if se.Cause == nil { + t.Error("the raw error was dropped; Diagnostics needs it") + } + return se +} diff --git a/gateway/kube/go.mod b/gateway/kube/go.mod index df86cb2..0e390d7 100644 --- a/gateway/kube/go.mod +++ b/gateway/kube/go.mod @@ -21,7 +21,7 @@ go 1.27.1 // // Raise these only to raise the SUPPORTED MINOR, and say why in the commit. require ( - github.com/ConfigButler/krm-stream/gateway v0.2.1 + github.com/ConfigButler/krm-stream/gateway v0.4.1-0.20261002130336-e723903f0c9e k8s.io/api v0.36.0 k8s.io/apimachinery v0.36.0 k8s.io/client-go v0.36.0 diff --git a/gateway/kube/go.sum b/gateway/kube/go.sum index bef48b9..34f5e52 100644 --- a/gateway/kube/go.sum +++ b/gateway/kube/go.sum @@ -1,5 +1,5 @@ -github.com/ConfigButler/krm-stream/gateway v0.2.1 h1:FqvODTWfNbkG4JcLG9rTKEXwZ0oDp3QPkIFLFSyFkZI= -github.com/ConfigButler/krm-stream/gateway v0.2.1/go.mod h1:BMdq7j79aNXEzaBept4sdM7cmCSWDm2hcK5G2IUJRjw= +github.com/ConfigButler/krm-stream/gateway v0.4.1-0.20261002130336-e723903f0c9e h1:uQVUgtSd29CIQ147h6XSfNSHOUZrdvVm9X4Q0iTo7eI= +github.com/ConfigButler/krm-stream/gateway v0.4.1-0.20261002130336-e723903f0c9e/go.mod h1:sVRhasWxXotQQHsyY4mKvvT0npkupDjrvg/i3Elq5+E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= diff --git a/gateway/observe.go b/gateway/observe.go index 20ca3fa..7b3006c 100644 --- a/gateway/observe.go +++ b/gateway/observe.go @@ -29,6 +29,9 @@ const ( ObservationConsumerResync ObservationKind = "consumer_resync" // ObservationSharedOverflow reports a shared-watch subscriber exceeding its bounded queue. ObservationSharedOverflow ObservationKind = "shared_overflow" + // ObservationRetryableError reports a logical stream ending with a non-terminal error the client + // must reconnect for, such as UPSTREAM_UNAVAILABLE, before attempting its frame. + ObservationRetryableError ObservationKind = "retryable_error" // ObservationTerminalError reports a logical-stream failure before attempting its // terminal frame. It does not guarantee delivery. ObservationTerminalError ObservationKind = "terminal_error" @@ -61,3 +64,23 @@ type ObserverFunc func(Observation) func (f ObserverFunc) Observe(observation Observation) { f(observation) } + +// Diagnostic is the raw error behind an error event, for the host's logs. Unlike an Observation it +// carries the error itself, which may name internal addresses, URLs or anything else an upstream put +// in it. That is why it has a hook of its own: the host decides what to log and what to redact. +type Diagnostic struct { + // Principal is nil when the caller could not be identified. + Principal Principal + // Scope is the zero Scope when the request was refused before a scope was parsed. + Scope Scope + // Code and Terminal are what the consumer was sent. + Code ErrorCode + Terminal bool + // Err is the complete error, including any StreamError.Cause. + Err error +} + +// Diagnostics receives a Diagnostic for every error event the gateway sends, and for every upstream +// error it recovers from with a new snapshot cycle. It runs on the stream goroutine and must return +// promptly. +type Diagnostics func(Diagnostic) diff --git a/gateway/reauthorize.go b/gateway/reauthorize.go index 7829243..acce24e 100644 --- a/gateway/reauthorize.go +++ b/gateway/reauthorize.go @@ -67,9 +67,11 @@ func (g *Gateway) authorizedCycle(ctx context.Context, principal Principal, scop return } if err != nil { - // Fail closed even if a host returns a normally recoverable StreamError. + // Fail closed even if a host returns a normally recoverable StreamError. The one + // exception is an unreachable authority (a SubjectAccessReview that could not be + // sent): the stream still ends, but the client may reconnect later. se := *asStreamError(err) - se.Terminal = true + se.Terminal = se.Code != CodeUpstreamUnavailable cancel() gate.mu.Unlock() done <- &se diff --git a/gateway/scope.go b/gateway/scope.go index d9921fb..5b15edc 100644 --- a/gateway/scope.go +++ b/gateway/scope.go @@ -176,6 +176,24 @@ type ScopePolicy struct { Targets []string // Resources are the group+resource pairs that may be streamed. Resources []GroupResource + // AnyResource leaves resource admission to the upstream's own authorization: any resource the + // scope syntax accepts passes, in a namespace or without one. Without one, the upstream decides + // what that means — the list of a cluster-scoped resource, or every namespace for a namespaced + // one, which Kubernetes RBAC allows only with a cluster-wide grant. + // + // It is another valid host policy, not a weaker allowlist: a host whose stream endpoint promises + // "whatever Kubernetes lets you read", as a `/k8s` proxy does, has deliberately no list to keep. + // The target allowlist, the label-selector switch, the refusal of API-server addresses and the + // Authorizer all still apply. + // + // What it does NOT do is check that the stream runs as the caller. That is the host's obligation, + // in its ClientFor, and no type check here could prove it: a kube.Backend can carry a service + // account's client, and a wrapper can hide a SharedBackend. With either, AnyResource streams every + // resource that identity can read to everyone your Authorizer admits. Use it only with a backend + // built from the caller's own credential. + // + // Setting both AnyResource and Resources is refused: Handler panics, and Validate admits nothing. + AnyResource bool // AllowLabelSelector permits a caller to narrow an already-allowed scope with Kubernetes label // selector syntax. The zero value refuses selectors rather than silently expanding the supported // request surface. A host that enables it should still constrain selector complexity at its edge. @@ -212,6 +230,12 @@ func (p ScopePolicy) Validate(s Scope) error { if !contains(p.Targets, s.Target) { return ScopeInvalid("target is not allowlisted: " + quoteOrEmpty(s.Target)) } + if p.AnyResource { + if len(p.Resources) > 0 { + return &StreamError{Code: CodeInternal, Terminal: true, Message: "scope policy sets both Resources and AnyResource"} + } + return nil + } for _, gr := range p.Resources { if gr.Group != s.Group || gr.Resource != s.Resource { continue diff --git a/gateway/seams.go b/gateway/seams.go index b67981a..0d28fd3 100644 --- a/gateway/seams.go +++ b/gateway/seams.go @@ -3,6 +3,7 @@ package gateway import ( "context" "fmt" + "time" ) // The seams. This file is the whole of CONTRIBUTING's one rule, expressed as Go: @@ -126,15 +127,28 @@ type WatchEvent struct { var ErrWatchClosed = fmt.Errorf("krm-stream: upstream watch closed") // StreamError is a protocol-level error the gateway emits to the consumer, and the type a host's -// Authorizer or ClientFor returns to choose the code the browser sees. +// Principal, Authorizer or ClientFor returns to choose the code the browser sees. +// +// Message goes on the wire, so it must be something the caller may read. Cause never does: it is the +// detail behind the code (a dial error, the URL that failed), and it reaches only the host's +// Diagnostics hook. type StreamError struct { Code ErrorCode Message string Terminal bool RetryAfterMs *int + Cause error } -func (e *StreamError) Error() string { return fmt.Sprintf("%s: %s", e.Code, e.Message) } +func (e *StreamError) Error() string { + if e.Cause != nil { + return fmt.Sprintf("%s: %s: %v", e.Code, e.Message, e.Cause) + } + return fmt.Sprintf("%s: %s", e.Code, e.Message) +} + +// Unwrap exposes Cause to errors.Is and errors.As. +func (e *StreamError) Unwrap() error { return e.Cause } // Event renders the error as the wire event. func (e *StreamError) Event() Event { @@ -158,6 +172,24 @@ func ScopeInvalid(msg string) *StreamError { return &StreamError{Code: CodeScopeInvalid, Message: msg, Terminal: true} } +// Unauthenticated refuses a caller whose credential is missing, expired or rejected. Terminal: signing +// in again is the remedy, and a page offers that rather than retrying. +func Unauthenticated(msg string) *StreamError { + return &StreamError{Code: CodeUnauthenticated, Message: msg, Terminal: true} +} + +// UpstreamUnavailable reports an upstream that cannot be reached right now. NOT terminal, but it ends +// the connection: the gateway sends it and closes, and the client reconnects with its own backoff, +// waiting at least retryAfter when that is positive (spec §4.3). Zero means "use your own backoff". +func UpstreamUnavailable(msg string, retryAfter time.Duration) *StreamError { + se := &StreamError{Code: CodeUpstreamUnavailable, Message: msg} + if retryAfter > 0 { + ms := int(retryAfter.Milliseconds()) + se.RetryAfterMs = &ms + } + return se +} + // ResyncRequired announces a loss of upstream continuity. NOT terminal: a fresh snapshot cycle // follows on the same connection, and the consumer converges (spec §5). func ResyncRequired(msg string) *StreamError { diff --git a/gateway/shared.go b/gateway/shared.go index 20d0c3b..cf1c88b 100644 --- a/gateway/shared.go +++ b/gateway/shared.go @@ -5,6 +5,7 @@ import ( "errors" "strings" "sync" + "time" ) // Fan-out: one upstream watch per SCOPE, not per browser tab. @@ -52,6 +53,17 @@ import ( // leak and never into a lie. const sharedQueueDepth = 256 +// The backoff after a shared upstream watch fails with UPSTREAM_UNAVAILABLE, whether opening it fails +// or the open watch dies of it. Its subscribers each reconnect on their own budget, so without it the +// API server would see one attempt per subscriber per retry. With it, a scope sees at most one +// attempt per backoff period however many subscribers are waiting, and the rest are told when to +// come back. Only a completed upstream snapshot resets it: a watch that opens and fails at once has +// not recovered. +const ( + sharedRetryMin = time.Second + sharedRetryMax = 30 * time.Second +) + // SharedOptions configures an opt-in SharedBackend. Zero values preserve the bounded, safe defaults. type SharedOptions struct { // QueueDepth is the maximum number of live events a subscriber may lag before it is resnapshotted. @@ -71,8 +83,16 @@ type SharedBackend struct { queueDepth int observer Observer - mu sync.Mutex - scopes map[string]*sharedScope + mu sync.Mutex + scopes map[string]*sharedScope + backoff map[string]*sharedBackoff + now func() time.Time +} + +// sharedBackoff is one scope's failed attempts to open its upstream watch. +type sharedBackoff struct { + until time.Time + delay time.Duration } // NewSharedBackend shares one upstream watch per scope across every consumer of it. @@ -94,6 +114,8 @@ func NewSharedBackendWithOptions(upstream Backend, options SharedOptions) *Share queueDepth: depth, observer: options.Observer, scopes: map[string]*sharedScope{}, + backoff: map[string]*sharedBackoff{}, + now: time.Now, } } @@ -118,9 +140,14 @@ func (b *SharedBackend) Watch(_ context.Context, scope Scope) (Watcher, error) { b.mu.Lock() s, ok := b.scopes[key] if !ok { + if wait := b.backoffRemainingLocked(key); wait > 0 { + b.mu.Unlock() + return nil, UpstreamUnavailable("the upstream is unavailable; retry later", wait) + } var err error s, err = b.startScope(scope, key) if err != nil { + b.recordFailureLocked(key, err) b.mu.Unlock() return nil, err } @@ -131,6 +158,43 @@ func (b *SharedBackend) Watch(_ context.Context, scope Scope) (Watcher, error) { return s.subscribe() } +func (b *SharedBackend) backoffRemainingLocked(key string) time.Duration { + if bo := b.backoff[key]; bo != nil { + return bo.until.Sub(b.now()) + } + return 0 +} + +// recordFailureLocked starts or doubles a scope's backoff after a retryable failure. Any other +// failure is not the upstream being away, and backing off would only delay the real answer; it +// leaves the backoff as it is, because it is not a recovery either. +func (b *SharedBackend) recordFailureLocked(key string, err error) { + var se *StreamError + if !errors.As(err, &se) || se == nil || se.Terminal || se.Code != CodeUpstreamUnavailable { + return + } + now := b.now() + for k, old := range b.backoff { + if now.Sub(old.until) > sharedRetryMax { // quiet for a full period: forget it, so the map stays small + delete(b.backoff, k) + } + } + bo := b.backoff[key] + if bo == nil { + bo = &sharedBackoff{delay: sharedRetryMin} + b.backoff[key] = bo + } else if bo.delay = 2 * bo.delay; bo.delay > sharedRetryMax { + bo.delay = sharedRetryMax + } + wait := bo.delay + if se.RetryAfterMs != nil { + if hint := time.Duration(*se.RetryAfterMs) * time.Millisecond; hint > wait { + wait = min(hint, sharedRetryMax) + } + } + bo.until = now.Add(wait) +} + // startScope opens the one upstream watch for a scope and pumps it. Called with b.mu held. // // The upstream watch gets its OWN context, deliberately detached from the subscriber whose arrival @@ -159,12 +223,27 @@ func (b *SharedBackend) startScope(scope Scope, key string) (*sharedScope, error return s, nil } -// forget drops a dead-or-empty scope so the next Watch opens a fresh one. -func (b *SharedBackend) forget(key string, s *sharedScope) { +// forget drops a dead-or-empty scope so the next Watch opens a fresh one. A retryable cause is +// recorded in the same critical section, so no Watch can slip in between and reopen the upstream +// without seeing the backoff. +func (b *SharedBackend) forget(key string, s *sharedScope, cause error) { b.mu.Lock() defer b.mu.Unlock() if b.scopes[key] == s { // not a newer incarnation delete(b.scopes, key) + b.recordFailureLocked(key, cause) + } +} + +// recovered resets a scope's backoff once its upstream has delivered a complete snapshot — but only +// if s is still the scope's current watch. The call happens after s.mu is released, and by then s +// may have been forgotten and a newer attempt may have failed; an obsolete watch's recovery must not +// erase that newer failure's backoff. +func (b *SharedBackend) recovered(key string, s *sharedScope) { + b.mu.Lock() + defer b.mu.Unlock() + if b.scopes[key] == s { + delete(b.backoff, key) } } @@ -334,12 +413,15 @@ func (s *sharedScope) pump(ctx context.Context) { return } + recovered := false s.mu.Lock() switch ev.Type { case WatchBookmark: if ev.InitialEventsEnd { // The upstream snapshot is complete: the cache is now a true picture of the scope, and - // everyone waiting for one can have it. + // everyone waiting for one can have it. It is also the upstream's recovery, so the + // backoff resets — after s.mu is released, since b.mu is never taken under it. + recovered = !s.synced s.synced = true for sub := range s.subs { if sub.awaiting { @@ -383,6 +465,9 @@ func (s *sharedScope) pump(ctx context.Context) { return } s.mu.Unlock() + if recovered { + s.backend.recovered(s.key, s) + } } } @@ -402,7 +487,7 @@ func (s *sharedScope) fanOutLocked(ev WatchEvent) { // die ends the scope: every subscriber is told continuity was lost, and the scope is removed so the // next Watch opens a fresh upstream. func (s *sharedScope) die(cause error) { - s.backend.forget(s.key, s) + s.backend.forget(s.key, s, cause) s.cancel() // Whatever ended the upstream — a clean close, a 410, a cancelled context — means one thing to a @@ -438,7 +523,7 @@ func (s *sharedScope) leave(sub *subscriber) { s.mu.Unlock() if empty { - s.backend.forget(s.key, s) + s.backend.forget(s.key, s, nil) s.cancel() // stops pump, which stops the upstream watch } } diff --git a/gateway/shared_test.go b/gateway/shared_test.go index a6e8558..7d6afec 100644 --- a/gateway/shared_test.go +++ b/gateway/shared_test.go @@ -411,3 +411,247 @@ func TestUpstreamErrorFansOutAndDoesNotStampede(t *testing.T) { t.Errorf("upstream opens = %d, want 2: five subscribers resyncing at once caused a stampede", up.opens()) } } + +// A shared upstream has no browser to hand its retry to, and each of its subscribers reconnects on +// its own budget. However many there are, the scope sees one open attempt per backoff period; the +// rest are told how long to wait. +func TestSharedBackendBacksOffAnUnavailableUpstream(t *testing.T) { + opens := 0 + up := backendFunc(func() (Watcher, error) { + opens++ + return nil, UpstreamUnavailable("the API server is unavailable", 0) + }) + b := NewSharedBackend(up) + now := time.Unix(0, 0) + b.now = func() time.Time { return now } + scope := Scope{Version: "v1", Resource: "configmaps", Namespace: "app"} + + retryAfter := func() int { + t.Helper() + _, err := b.Watch(t.Context(), scope) + var se *StreamError + if !errors.As(err, &se) || se.Code != CodeUpstreamUnavailable || se.Terminal { + t.Fatalf("Watch() = %v, want UPSTREAM_UNAVAILABLE", err) + } + if se.RetryAfterMs == nil { + return 0 + } + return *se.RetryAfterMs + } + + retryAfter() + for range 50 { // a reconnect storm within the first second + if ms := retryAfter(); ms <= 0 || ms > 1000 { + t.Fatalf("retryAfterMs = %d, want the remaining backoff", ms) + } + } + if opens != 1 { + t.Fatalf("upstream opens = %d, want 1 during the backoff", opens) + } + + now = now.Add(time.Second) + retryAfter() + if opens != 2 { + t.Fatalf("upstream opens = %d, want a second attempt once the backoff expired", opens) + } + if ms := retryAfter(); ms != 2000 { + t.Errorf("retryAfterMs = %d, want the backoff doubled to 2000", ms) + } +} + +// dyingWatcher opens fine and then fails at once: through Next, or as a WatchError event. +type dyingWatcher struct { + err error + asEvent bool +} + +func (w dyingWatcher) Next(context.Context) (WatchEvent, error) { + if w.asEvent { + return WatchEvent{Type: WatchError, Err: w.err}, nil + } + return WatchEvent{}, w.err +} +func (dyingWatcher) Stop() {} + +// A watch that opens and then dies of an unavailable upstream is the same outage as one that fails +// to open, and must be bounded the same way: one upstream attempt per backoff period, however many +// subscribers reconnect at once, honouring the upstream's hint. +func TestSharedBackendBacksOffAWatchThatDiesAfterOpening(t *testing.T) { + for _, asEvent := range []bool{false, true} { + t.Run(map[bool]string{false: "Next error", true: "WatchError event"}[asEvent], func(t *testing.T) { + var mu sync.Mutex + opens := 0 + up := backendFunc(func() (Watcher, error) { + mu.Lock() + opens++ + mu.Unlock() + return dyingWatcher{err: UpstreamUnavailable("the API server is unavailable", 10*time.Second), asEvent: asEvent}, nil + }) + b := NewSharedBackend(up) + var clock sync.Mutex + now := time.Unix(0, 0) + b.now = func() time.Time { clock.Lock(); defer clock.Unlock(); return now } + scope := Scope{Version: "v1", Resource: "configmaps", Namespace: "app"} + + // What a subscriber's stream loop does: resync in band on a closed watch, and give up on + // anything else (the gateway sends it and closes). + reconnect := func() *StreamError { + for { + w, err := b.Watch(t.Context(), scope) + if err == nil { + for err == nil { + var ev WatchEvent + ev, err = w.Next(t.Context()) + if err == nil && ev.Type == WatchError { + err = ev.Err + } + } + w.Stop() + } + var se *StreamError + if errors.As(err, &se) && se.Code == CodeUpstreamUnavailable { + return se + } + if !errors.Is(err, ErrWatchClosed) && (se == nil || se.Code != CodeResyncRequired) { + t.Errorf("unexpected error: %v", err) + return nil + } + } + } + + var wg sync.WaitGroup + hints := make(chan int, 100) + for range 100 { + wg.Go(func() { + if se := reconnect(); se != nil && se.RetryAfterMs != nil { + hints <- *se.RetryAfterMs + } + }) + } + wg.Wait() + close(hints) + if opens != 1 { + t.Fatalf("upstream opens = %d, want 1 for 100 concurrent reconnects", opens) + } + for ms := range hints { + if ms <= 0 || ms > 10_000 { + t.Fatalf("retryAfterMs = %d, want the upstream's 10s hint or what remains of it", ms) + } + } + + clock.Lock() + now = now.Add(9 * time.Second) + clock.Unlock() + _ = reconnect() + if opens != 1 { + t.Fatalf("upstream opens = %d, want the 10s hint honoured", opens) + } + clock.Lock() + now = now.Add(time.Second) + clock.Unlock() + _ = reconnect() + if opens != 2 { + t.Fatalf("upstream opens = %d, want one attempt once the hint expired", opens) + } + }) + } +} + +// Obtaining a watcher is not recovery; a completed upstream snapshot is. Only then does the backoff +// reset, so a flapping upstream keeps doubling it. +func TestSharedBackoffResetsOnlyAfterASnapshot(t *testing.T) { + healthy := false + up := backendFunc(func() (Watcher, error) { + if healthy { + return &stubWatcher{events: []WatchEvent{{Type: WatchBookmark, InitialEventsEnd: true}}}, nil + } + return dyingWatcher{err: UpstreamUnavailable("the API server is unavailable", 0)}, nil + }) + b := NewSharedBackend(up) + now := time.Unix(0, 0) + b.now = func() time.Time { return now } + scope := Scope{Version: "v1", Resource: "configmaps", Namespace: "app"} + key := scopeKey(scope) + + for i := range 3 { + w, err := b.Watch(t.Context(), scope) + if err != nil { + t.Fatalf("attempt %d: %v", i, err) + } + for err == nil { + _, err = w.Next(t.Context()) + } + w.Stop() + b.mu.Lock() + delay := b.backoff[key].delay + b.mu.Unlock() + if want := sharedRetryMin << i; delay != want { + t.Fatalf("attempt %d: backoff = %v, want %v: an open that fails at once is not recovery", i, delay, want) + } + now = now.Add(delay) + } + + healthy = true + w, err := b.Watch(t.Context(), scope) + if err != nil { + t.Fatal(err) + } + defer w.Stop() + if ev := next(t, w); ev.Type != WatchBookmark || !ev.InitialEventsEnd { + t.Fatalf("got %+v, want the snapshot boundary", ev) + } + b.mu.Lock() + _, backingOff := b.backoff[key] + b.mu.Unlock() + if backingOff { + t.Error("the backoff survived a completed snapshot") + } +} + +// A watch's recovery is reported after it releases its own lock, so it can arrive late: after the +// watch was forgotten and a replacement failed. That stale report must not erase the replacement's +// backoff and let the next subscriber reopen the upstream at once. +func TestAnObsoleteWatchsRecoveryCannotEraseANewerBackoff(t *testing.T) { + opens := 0 + failing := false + up := backendFunc(func() (Watcher, error) { + opens++ + if failing { + return nil, UpstreamUnavailable("the API server is unavailable", 0) + } + return &stubWatcher{}, nil // opens, then idles: its recovery is replayed by hand below + }) + b := NewSharedBackend(up) + now := time.Unix(0, 0) + b.now = func() time.Time { return now } + scope := Scope{Version: "v1", Resource: "configmaps", Namespace: "app"} + key := scopeKey(scope) + + // 1. The old watch is open, and is about to report its recovery. + w, err := b.Watch(t.Context(), scope) + if err != nil { + t.Fatal(err) + } + b.mu.Lock() + old := b.scopes[key] + b.mu.Unlock() + + // 2. Its last subscriber leaves, so it is forgotten. + w.Stop() + + // 3. A replacement fails to open and records a backoff. + failing = true + if _, err := b.Watch(t.Context(), scope); err == nil { + t.Fatal("the replacement opened") + } + + // 4. The old watch's delayed recovery arrives. + b.recovered(key, old) + + if _, err := b.Watch(t.Context(), scope); err == nil { + t.Fatal("a backoff-limited Watch succeeded") + } + if opens != 2 { + t.Errorf("upstream opens = %d, want 2: the stale recovery erased the newer backoff", opens) + } +} diff --git a/gateway/stream.go b/gateway/stream.go index 11db932..0857310 100644 --- a/gateway/stream.go +++ b/gateway/stream.go @@ -87,6 +87,9 @@ type Gateway struct { Projections ProjectionPolicy // Observer receives low-cardinality lifecycle signals. Nil disables observations. Observer Observer + // Diagnostics receives the raw error behind every error event, including detail kept off the + // wire. Nil discards it. + Diagnostics Diagnostics // HeartbeatInterval controls SSE heartbeats when ServeStream is used. Zero uses the package // default. It has no effect on the transport-neutral Stream method. HeartbeatInterval time.Duration @@ -106,11 +109,11 @@ type Gateway struct { Ordering ResourceVersionOrdering } -// Stream runs one consumer's stream until the context is done or a terminal error is emitted. +// Stream runs one consumer's stream until the context is done or an error event ends it. // -// It returns after emitting a terminal error; the caller (the SSE handler) must then CLOSE the -// connection. A browser's EventSource reconnects automatically otherwise, and would hammer a -// forbidden scope forever. +// It returns after emitting a terminal error, or a non-terminal one such as UPSTREAM_UNAVAILABLE +// that the client must reconnect for; the caller (the SSE handler) must then CLOSE the connection. +// Only RESYNC_REQUIRED is recovered on the same connection. func (g *Gateway) Stream(ctx context.Context, principal Principal, scope Scope, sink Sink) error { return g.StreamProjection(ctx, principal, scope, "", sink) } @@ -153,19 +156,19 @@ func (g *Gateway) StreamProjection(ctx context.Context, principal Principal, sco // hammering a forbidden scope forever. // if err := g.Auth.Authorize(ctx, principal, scope); err != nil { - return g.emitTerminal(ctx, sink, scope, "", err) + return g.emitError(ctx, sink, principal, scope, "", err) } projection, err := policy.SelectProjection(ctx, principal, scope, requested) if err != nil { - return g.emitTerminal(ctx, sink, scope, "", err) + return g.emitError(ctx, sink, principal, scope, "", err) } if !isBuiltinProjection(projection) { - return g.emitTerminal(ctx, sink, scope, projection, &StreamError{Code: CodeInternal, Terminal: true, Message: "projection policy selected an unknown projection: " + string(projection)}) + return g.emitError(ctx, sink, principal, scope, projection, &StreamError{Code: CodeInternal, Terminal: true, Message: "projection policy selected an unknown projection: " + string(projection)}) } g.observe(Observation{Kind: ObservationCycleStarted, Scope: scope, Projection: projection}) backend, err := g.Clients(ctx, scope.Target, principal) if err != nil { - return g.emitTerminal(ctx, sink, scope, projection, err) + return g.emitError(ctx, sink, principal, scope, projection, err) } err = g.authorizedCycle(ctx, principal, scope, requested, policy, backend, projection, revisions, sink) @@ -182,11 +185,17 @@ func (g *Gateway) StreamProjection(ctx context.Context, principal Principal, sco continue } + // Continuity lost (a 410, a partial object, an overflowing shared queue): announce, + // resnapshot, carry on. Every other error ends the connection, including retryable ones. The + // client owns the retry: it already has a bounded budget, backoff and jitter, a second backoff + // here would multiply with it, and a held connection would pin a goroutine and a subscription + // for an outage of unknown length. A reconnect costs the same fresh snapshot a retry here would. var se *StreamError - if errors.As(err, &se) && !se.Terminal { - continue // a recoverable upstream error: announce, resnapshot, carry on + if errors.As(err, &se) && se != nil && !se.Terminal && se.Code == CodeResyncRequired { + g.diagnose(Diagnostic{Principal: principal, Scope: scope, Code: se.Code, Err: err}) + continue } - return g.emitTerminal(ctx, sink, scope, projection, err) + return g.emitError(ctx, sink, principal, scope, projection, err) } } @@ -329,11 +338,22 @@ func (g *Gateway) observe(observation Observation) { } } -func (g *Gateway) emitTerminal(ctx context.Context, sink Sink, scope Scope, projection Projection, err error) error { +func (g *Gateway) diagnose(d Diagnostic) { + if g.Diagnostics != nil { + g.Diagnostics(d) + } +} + +// emitError sends the error event that ends a stream. The caller closes the connection after it, +// whether or not it is terminal. +func (g *Gateway) emitError(ctx context.Context, sink Sink, principal Principal, scope Scope, projection Projection, err error) error { se := asStreamError(err) + g.diagnose(Diagnostic{Principal: principal, Scope: scope, Code: se.Code, Terminal: se.Terminal, Err: err}) + kind := ObservationRetryableError if se.Terminal { - g.observe(Observation{Kind: ObservationTerminalError, Scope: scope, Projection: projection, Code: se.Code}) + kind = ObservationTerminalError } + g.observe(Observation{Kind: kind, Scope: scope, Projection: projection, Code: se.Code}) if emitErr := sink.Emit(ctx, se.Event()); emitErr != nil { return emitErr } @@ -505,6 +525,11 @@ func isDecimalResourceVersion(rv string) bool { // asStreamError finds the *StreamError a host meant to send, or makes one. // +// A made one says "internal error" and nothing else. Whatever an unexpected error says (a dial error +// naming the cluster's internal address, a URL a backend promised never to show a browser) is for the +// host's Diagnostics hook, not the wire. A StreamError the host or a backend chose keeps its message: +// choosing a message is choosing to send it. +// // The `se == nil` arm is the typed-nil trap seen from the other side. A host whose Authorizer is // declared to return `error` and returns a nil *StreamError hands us a non-nil error interface // wrapping nothing: errors.As matches, sets se to nil, and every path downstream would dereference it @@ -513,7 +538,7 @@ func isDecimalResourceVersion(rv string) bool { func asStreamError(err error) *StreamError { var se *StreamError if !errors.As(err, &se) || se == nil { - return &StreamError{Code: CodeInternal, Message: err.Error(), Terminal: true} + return &StreamError{Code: CodeInternal, Message: "internal error", Terminal: true, Cause: err} } return se } diff --git a/gateway/stream_test.go b/gateway/stream_test.go index 32551e2..3612e0a 100644 --- a/gateway/stream_test.go +++ b/gateway/stream_test.go @@ -675,3 +675,55 @@ func TestFinalBookkeepingSuppressionAcrossProjections(t *testing.T) { }) } } + +// RESYNC_REQUIRED recovers on the same connection; UPSTREAM_UNAVAILABLE does not. It is sent with its +// code and hint, then the stream ends so the client reconnects on its own budget. Both reach +// Diagnostics with the raw cause. +func TestOnlyContinuityLossIsRecoveredInBand(t *testing.T) { + var diagnostics []Diagnostic + var observations []Observation + watches := 0 + gw := &Gateway{ + Auth: AllowAll{}, + Diagnostics: func(d Diagnostic) { diagnostics = append(diagnostics, d) }, + Observer: ObserverFunc(func(o Observation) { observations = append(observations, o) }), + Clients: func(context.Context, string, Principal) (Backend, error) { + return backendFunc(func() (Watcher, error) { + watches++ + if watches == 1 { + return &stubWatcher{events: []WatchEvent{ + {Type: WatchBookmark, InitialEventsEnd: true}, + {Type: WatchError, Err: ResyncRequired("expired")}, + }}, nil + } + se := UpstreamUnavailable("the API server is unavailable", 0) + se.Cause = errors.New("dial tcp 10.43.0.1:443: connect: connection refused") + return nil, se + }), nil + }, + } + sink := &recordingSink{} + err := gw.Stream(t.Context(), nil, Scope{Version: "v1", Resource: "configmaps"}, sink) + + var se *StreamError + if !errors.As(err, &se) || se.Code != CodeUpstreamUnavailable || se.Terminal { + t.Fatalf("Stream() = %v, want a non-terminal UPSTREAM_UNAVAILABLE", err) + } + if watches != 2 { + t.Errorf("watches = %d, want 2: one resync, then the unavailable upstream ends the stream", watches) + } + last := sink.events[len(sink.events)-1] + if last.Code != CodeUpstreamUnavailable || last.Terminal || strings.Contains(last.Message, "10.43") { + t.Errorf("last event = %+v", last) + } + if len(diagnostics) != 2 || diagnostics[0].Code != CodeResyncRequired || !strings.Contains(diagnostics[1].Err.Error(), "10.43.0.1") { + t.Errorf("diagnostics = %+v", diagnostics) + } + if !sawObservation(observations, ObservationRetryableError, "") || sawObservation(observations, ObservationTerminalError, "") { + t.Errorf("observations = %+v, want a retryable error and no terminal one", observations) + } +} + +type backendFunc func() (Watcher, error) + +func (f backendFunc) Watch(context.Context, Scope) (Watcher, error) { return f() } diff --git a/packages/krm-stream/CHANGELOG.md b/packages/krm-stream/CHANGELOG.md index 84e7e7d..89666bb 100644 --- a/packages/krm-stream/CHANGELOG.md +++ b/packages/krm-stream/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## Unreleased + +- The managed connector waits at least the server's retry hint (HTTP `Retry-After`, or an error + event's `retryAfterMs`) before reconnecting, within `maxRetryDelayMs`. A completed snapshot + discards the hint. `onError` receives it as a fourth argument. +- A refusal whose body is a Kubernetes `Status` reports its `message` instead of `stream: HTTP + `. The body read is bounded to 16 KiB and 2 seconds, and stops when the stream closes. +- HTTP 429, 502, 503 and 504 report `UPSTREAM_UNAVAILABLE` rather than `INTERNAL`. + + ## [0.4.0](https://github.com/ConfigButler/krm-stream/compare/@configbutler/krm-stream-v0.3.0...@configbutler/krm-stream-v0.4.0) (2026-09-11) diff --git a/packages/krm-stream/README.md b/packages/krm-stream/README.md index 43c5d5e..fa2bacc 100644 --- a/packages/krm-stream/README.md +++ b/packages/krm-stream/README.md @@ -67,7 +67,8 @@ await connection.closed; Import `connectManagedResourceStream` from the package. It uses fetch for same-origin cookies or bearer headers; `credentials: "include"` opts into cross-origin cookies. It requests a fresh snapshot -on sequence gaps, network failures, HTTP 408/429/5xx and EOF. Existing drafts survive recovery. +on sequence gaps, network failures, HTTP 408/429/5xx, EOF, and a connection the gateway closes after +a retryable error such as `UPSTREAM_UNAVAILABLE`. Existing drafts survive recovery. States are `connecting`, `syncing`, `live`, `retrying`, `closed`, `terminal`, and `exhausted`. A reset makes the connection `syncing` until `synced`; enable saves while `live`. @@ -77,7 +78,14 @@ this threshold). Brief snapshots do not replenish the budget; reset, disconnect the health timer. Terminal protocol errors and HTTP client errors (including 401/403, excluding 408/429) stop retries. `close()` or `signal` cancels the stream and pending backoff; `closed` resolves after cleanup. Create a new handle after credentials change or an -explicit user retry. The low-level fetch and native EventSource connectors remain available; native +explicit user retry. + +The server's retry hint sets the least the next reconnect waits, within `maxRetryDelayMs`: an HTTP +`Retry-After`, or an error event's `retryAfterMs`. `onError` receives it as its fourth argument. When +a refusal's body is a Kubernetes `Status`, as a host proxying `/k8s` might send, `onError` receives +its `message` instead of `stream: HTTP 403`. + +The low-level fetch and native EventSource connectors remain available; native EventSource owns network reconnects but closes on sequence gaps. Use the managed connector for bounded recovery and observable lifecycle state. diff --git a/packages/krm-stream/src/connection.ts b/packages/krm-stream/src/connection.ts index 6c30fdc..f3411d8 100644 --- a/packages/krm-stream/src/connection.ts +++ b/packages/krm-stream/src/connection.ts @@ -28,7 +28,13 @@ export interface ManagedStreamHandle extends StreamHandle { /** Managed fetch transport for both session cookies and bearer headers. Every reconnect requests a * fresh snapshot; the existing store and its drafts survive. HTTP 401/403 and terminal protocol - * errors stop permanently. EOF, network failures and sequence gaps consume a bounded retry budget. */ + * errors stop permanently. EOF, network failures, sequence gaps and retryable errors such as + * UPSTREAM_UNAVAILABLE consume a bounded retry budget. + * + * A server's retry hint (an HTTP `Retry-After`, or an error event's `retryAfterMs`) sets the least + * the next reconnect waits, within `maxRetryDelayMs`. A non-terminal event on an open connection does + * not itself reconnect: the gateway recovers RESYNC_REQUIRED in band, and closes the connection + * after any error it wants the client to retry. A completed snapshot discards the hint. */ export function connectManagedResourceStream( url: string, store: LiveResourceStore, @@ -56,6 +62,7 @@ export function connectManagedResourceStream( const subscribers = new Set<(state: Readonly) => void>(); let state: Readonly = Object.freeze({ status: "connecting", retries: 0 }); let terminal = false; + let hintMs: number | undefined; let healthTimer: ReturnType | undefined; const clearHealthTimer = () => { clearTimeout(healthTimer); @@ -79,6 +86,7 @@ export function connectManagedResourceStream( while (!controller.signal.aborted) { publish("connecting"); if (controller.signal.aborted) break; + hintMs = undefined; const stream = connectResourceStream(url, store, { ...opts, signal: controller.signal, @@ -98,6 +106,7 @@ export function connectManagedResourceStream( opts.onChange?.(change); }, onSynced: () => { + hintMs = undefined; if (state.status !== "live") { healthTimer = setTimeout(() => { healthTimer = undefined; @@ -110,10 +119,11 @@ export function connectManagedResourceStream( publish("live"); opts.onSynced?.(); }, - onError: (code, message, isTerminal) => { + onError: (code, message, isTerminal, retryAfterMs) => { if (isTerminal) clearHealthTimer(); terminal ||= isTerminal; - opts.onError?.(code, message, isTerminal); + if (!isTerminal && retryAfterMs !== undefined) hintMs = retryAfterMs; + opts.onError?.(code, message, isTerminal, retryAfterMs); }, }); await stream.closed; @@ -128,7 +138,8 @@ export function connectManagedResourceStream( return; } const ceiling = Math.min(cap, delay * 2 ** Math.min(state.retries, 30)); - const wait = Math.floor(ceiling * (0.5 + Math.random() * 0.5)); + const jittered = Math.floor(ceiling * (0.5 + Math.random() * 0.5)); + const wait = Math.min(cap, Math.max(jittered, hintMs ?? 0)); state = { ...state, retries: state.retries + 1 }; publish("retrying", wait); await new Promise((resolve) => { diff --git a/packages/krm-stream/src/sse.ts b/packages/krm-stream/src/sse.ts index 9fb887f..0c2c7a8 100644 --- a/packages/krm-stream/src/sse.ts +++ b/packages/krm-stream/src/sse.ts @@ -149,8 +149,10 @@ export interface StreamOptions { onOpen?: () => void; /** Defaults to same-origin; use include for a cross-origin cookie gateway. */ credentials?: RequestCredentials; - /** Called for protocol and HTTP errors. A terminal error ends the connection without retry. */ - onError?: (code: ErrorCode, message: string, terminal: boolean) => void; + /** Called for protocol and HTTP errors. A terminal error ends the connection without retry. + * `retryAfterMs` is the server's hint for a retryable error: an error event's `retryAfterMs`, or + * an HTTP `Retry-After`. */ + onError?: (code: ErrorCode, message: string, terminal: boolean, retryAfterMs?: number) => void; /** Called at the end of every snapshot cycle. The store is now consistent: a good moment to paint. */ onSynced?: () => void; /** Called after any change, with what the event did and which resource it did it to. */ @@ -188,10 +190,18 @@ export function connectResourceStream(url: string, store: LiveResourceStore, opt credentials: opts.credentials ?? "same-origin", }); if (!res.ok || !res.body) { - const code = res.status === 401 ? "UNAUTHENTICATED" : res.status === 403 ? "FORBIDDEN" : "INTERNAL"; + const code: ErrorCode = + res.status === 401 + ? "UNAUTHENTICATED" + : res.status === 403 + ? "FORBIDDEN" + : res.status === 429 || res.status === 502 || res.status === 503 || res.status === 504 + ? "UPSTREAM_UNAVAILABLE" + : "INTERNAL"; const terminal = res.status >= 400 && res.status < 500 && res.status !== 408 && res.status !== 429; - await res.body?.cancel(); - opts.onError?.(code, `stream: HTTP ${res.status}`, terminal); + const message = (await statusMessage(res, controller.signal)) ?? `stream: HTTP ${res.status}`; + if (controller.signal.aborted) return; // closed while reading the refusal: report nothing + opts.onError?.(code, message, terminal, retryAfter(res.headers.get("Retry-After"))); return; } @@ -234,6 +244,73 @@ export function connectResourceStream(url: string, store: LiveResourceStore, opt }; } +/** The largest refusal body read in search of a Kubernetes Status message. */ +const maxStatusBytes = 16 * 1024; +/** How long a refusal body may take to arrive. A known 401, 403 or 429 must not wait on a server + * that stalls the body explaining it. */ +const statusBudgetMs = 2_000; + +/** The `message` of a Kubernetes `Status` body, as a host proxying `/k8s` refuses with, or + * undefined. It reads at most maxStatusBytes for at most budgetMs, stops when `signal` aborts, and + * always releases the body. */ +export async function statusMessage( + res: Response, + signal: AbortSignal, + budgetMs = statusBudgetMs, +): Promise { + if (!res.body) return undefined; + if (signal.aborted || !/^application\/json\b/i.test(res.headers.get("Content-Type") ?? "")) { + await res.body.cancel().catch(() => {}); + return undefined; + } + const reader = res.body.getReader(); + let gaveUp = false; + // Cancelling the reader settles a pending read(), so a quiet body cannot hold this open. + const giveUp = () => { + gaveUp = true; + void reader.cancel().catch(() => {}); + }; + const timer = setTimeout(giveUp, budgetMs); + signal.addEventListener("abort", giveUp, { once: true }); + const chunks: Uint8Array[] = []; + let size = 0; + try { + for (;;) { + const { done, value } = await reader.read(); + if (gaveUp) return undefined; + if (done) break; + size += value.byteLength; + if (size > maxStatusBytes) return undefined; + chunks.push(value); + } + const bytes = new Uint8Array(size); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + const status: unknown = JSON.parse(new TextDecoder().decode(bytes)); + if (typeof status !== "object" || status === null) return undefined; + const { kind, message } = status as { kind?: unknown; message?: unknown }; + return kind === "Status" && typeof message === "string" && message !== "" ? message : undefined; + } catch { + return undefined; + } finally { + clearTimeout(timer); + signal.removeEventListener("abort", giveUp); + await reader.cancel().catch(() => {}); + } +} + +/** An HTTP `Retry-After` in milliseconds: delay-seconds or an HTTP-date. */ +export function retryAfter(header: string | null, now = Date.now()): number | undefined { + if (header === null) return undefined; + const value = header.trim(); + if (/^\d+$/.test(value)) return Number(value) * 1000; + const at = Date.parse(value); + return Number.isNaN(at) ? undefined : Math.max(0, at - now); +} + /** Consume a resource stream with the browser's native EventSource. This is the same-origin * session-cookie path — the baseline a v1 gateway MUST support, because a cookie is the only * credential EventSource can carry. */ @@ -294,7 +371,8 @@ function feed(store: LiveResourceStore, sequence: StreamSequence, ev: StreamEven return true; } if (ev.type === "error") { - opts.onError?.(ev.code ?? "INTERNAL", ev.message ?? "", ev.terminal ?? false); + const hint = typeof ev.retryAfterMs === "number" && ev.retryAfterMs >= 0 ? ev.retryAfterMs : undefined; + opts.onError?.(ev.code ?? "INTERNAL", ev.message ?? "", ev.terminal ?? false, hint); return ev.terminal === true; } const change = applyStreamEvent(store, ev); diff --git a/packages/krm-stream/test/connection.test.ts b/packages/krm-stream/test/connection.test.ts index 28dbba2..fec1bfc 100644 --- a/packages/krm-stream/test/connection.test.ts +++ b/packages/krm-stream/test/connection.test.ts @@ -271,3 +271,144 @@ test("snapshot resets restart the health interval and close removes the timer", t.mock.timers.tick(100); assert.equal(handle.state.status, "closed"); }); + +/** Runs a managed stream until its first retry is scheduled, and returns how long it would wait. */ +async function firstRetryDelay(fetch: typeof globalThis.fetch, opts: { maxRetryDelayMs?: number } = {}) { + let delay: number | undefined; + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + retryDelayMs: 0, + ...opts, + fetch, + onStateChange: (s) => { + if (s.status === "retrying" && delay === undefined) { + delay = s.retryInMs; + handle.close(); + } + }, + }); + await handle.closed; + return delay; +} + +test("a retryable error's retryAfterMs sets the least the reconnect waits", async () => { + const errors: unknown[][] = []; + const delay = await firstRetryDelay(async () => + response([{ seq: 1, type: "error", code: "UPSTREAM_UNAVAILABLE", terminal: false, retryAfterMs: 1500 }]), + ); + assert.equal(delay, 1500); + + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + maxRetries: 0, + fetch: async () => + response([{ seq: 1, type: "error", code: "UPSTREAM_UNAVAILABLE", terminal: false, retryAfterMs: 1500 }]), + onError: (...args) => errors.push(args), + }); + await handle.closed; + assert.deepEqual(errors, [["UPSTREAM_UNAVAILABLE", "", false, 1500]]); +}); + +test("HTTP Retry-After is honoured within the client's own cap", async () => { + const fetch = async () => new Response(null, { status: 429, headers: { "Retry-After": "2" } }); + assert.equal(await firstRetryDelay(fetch), 2000); + assert.equal(await firstRetryDelay(fetch, { maxRetryDelayMs: 300 }), 300); +}); + +test("a hint is discarded once a snapshot completes on the same connection", async () => { + const delay = await firstRetryDelay(async () => + response([ + { seq: 1, type: "error", code: "RESYNC_REQUIRED", terminal: false, retryAfterMs: 5000 }, + { seq: 2, type: "reset" }, + { seq: 3, type: "synced" }, + ]), + ); + assert.equal(delay, 0); +}); + +test("a refusal's Kubernetes Status message is shown instead of the bare status", async () => { + const messages: string[] = []; + const refuse = + (body: string, contentType = "application/json") => + async () => + new Response(body, { status: 403, headers: { "Content-Type": contentType } }); + for (const fetch of [ + refuse(JSON.stringify({ kind: "Status", code: 403, message: 'notes is forbidden: User "carol" cannot watch' })), + refuse(JSON.stringify({ kind: "Status", message: "not json" }), "text/plain"), + refuse(JSON.stringify({ kind: "Status", message: "x".repeat(20_000) })), + refuse("{not json"), + ]) { + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + fetch, + onError: (_code, message) => messages.push(message), + }); + await handle.closed; + } + assert.deepEqual(messages, [ + 'notes is forbidden: User "carol" cannot watch', + "stream: HTTP 403", + "stream: HTTP 403", + "stream: HTTP 403", + ]); +}); + +/** A JSON refusal whose body sends `prefix` and then goes quiet. Records whether it was cancelled. */ +function stalledRefusal(prefix = "") { + const body = { cancelled: false }; + const response = new Response( + new ReadableStream({ + start(controller) { + if (prefix) controller.enqueue(new TextEncoder().encode(prefix)); + }, + cancel() { + body.cancelled = true; + }, + }), + { status: 403, headers: { "Content-Type": "application/json" } }, + ); + return { response, body }; +} + +test("close() while a refusal body is quiet ends the stream and cancels the body", async () => { + const { response, body } = stalledRefusal(); + const errors: string[] = []; + let fetched!: () => void; + const wasFetched = new Promise((resolve) => { + fetched = resolve; + }); + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + fetch: async () => { + fetched(); + return response; + }, + onError: (_code, message) => errors.push(message), + }); + await wasFetched; + await new Promise((resolve) => setTimeout(resolve, 10)); + handle.close(); + await handle.closed; + assert.equal(handle.state.status, "closed"); + assert.equal(body.cancelled, true); + assert.deepEqual(errors, [], "a closed stream reports nothing"); +}); + +for (const [name, prefix] of [ + ["quiet", ""], + ["partial", '{"kind":"Status","mess'], +] as const) + test(`a ${name} refusal body falls back to the bare status within its budget`, async () => { + const { statusMessage } = await import("../src/sse.ts"); + const { response, body } = stalledRefusal(prefix); + const started = Date.now(); + assert.equal(await statusMessage(response, new AbortController().signal, 20), undefined); + assert.ok(Date.now() - started < 1_000, "the budget bounds the read"); + assert.equal(body.cancelled, true); + }); + +test("aborting during a refusal-body read stops it at once", async () => { + const { statusMessage } = await import("../src/sse.ts"); + const { response, body } = stalledRefusal('{"kind":'); + const abort = new AbortController(); + const read = statusMessage(response, abort.signal, 60_000); + setTimeout(() => abort.abort(), 10); + assert.equal(await read, undefined); + assert.equal(body.cancelled, true); +}); diff --git a/packages/krm-stream/test/wire.test.ts b/packages/krm-stream/test/wire.test.ts index c008ad0..f874498 100644 --- a/packages/krm-stream/test/wire.test.ts +++ b/packages/krm-stream/test/wire.test.ts @@ -182,3 +182,13 @@ function decodeInChunks(bytes: string, size: number): StreamEvent[] { } return out; } + +test("Retry-After accepts delay-seconds and an HTTP-date", async () => { + const { retryAfter } = await import("../src/sse.ts"); + const now = Date.parse("2026-10-02T12:00:00Z"); + assert.equal(retryAfter("3", now), 3000); + assert.equal(retryAfter("Fri, 02 Oct 2026 12:00:05 GMT", now), 5000); + assert.equal(retryAfter("Fri, 02 Oct 2026 11:00:00 GMT", now), 0); + assert.equal(retryAfter("soon", now), undefined); + assert.equal(retryAfter(null, now), undefined); +}); diff --git a/spec/v1.md b/spec/v1.md index bbe0bd5..af4f573 100644 --- a/spec/v1.md +++ b/spec/v1.md @@ -254,6 +254,27 @@ stable client behavior. v1 codes: reconnects automatically otherwise and will hammer a forbidden scope forever. - `retryAfterMs` is advice for a non-terminal error; `null`/absent means "use your own backoff". +**Who recovers.** The two non-terminal codes recover differently, and the difference is normative: + +- `RESYNC_REQUIRED` is recovered by the **gateway, on the same connection**: a new snapshot cycle + follows it immediately (§5). +- Any other non-terminal error (`UPSTREAM_UNAVAILABLE`, a non-terminal `INTERNAL`) is recovered by + the **client, on a new connection**. The gateway MUST send it with its `code` and `retryAfterMs` + intact, and then close the connection. It MUST NOT retry such an error itself on the open + connection. One retry policy, the client's, keeps two backoffs from multiplying, and a closed + connection holds nothing on the server for an outage of unknown length. A reconnect costs the same + fresh snapshot a retry in the gateway would. +- A client reconnecting after a non-terminal error SHOULD wait at least `retryAfterMs`, or an HTTP + `Retry-After` on a refused request, within its own maximum delay. A non-terminal event on a + connection that stays open is not by itself a reason to reconnect. +- A gateway that shares one upstream watch among many clients MUST bound its own attempts to reopen + an unavailable upstream; it cannot rely on the clients' backoff alone, because their attempts add up. + +An error that is not one the gateway expected MUST NOT carry the upstream's error text in `message`: +it may name internal addresses, URLs or credentials. The gateway sends a generic message and gives +the detail to its host. A message the host or the upstream's authorization chose for the caller, +such as Kubernetes' explanation of a 403, is not such text and may be sent. + ## 5. Snapshot cycles — `reset` … `synced` A **snapshot cycle** is `reset`, zero or more `added`, then `synced`. Between `synced` and the next @@ -407,8 +428,10 @@ object RV nor `seq` provides downstream resume; every new connection requires a ## 8. Scope & the request A stream is opened for a **normalized, server-validated scope**, never a raw Kubernetes URL. The -browser asks for a *logical* scope; the server maps it to an allowlisted target + GVR + the caller's -permitted namespace, and denies by default. The canonical scope is: +browser asks for a *logical* scope; the server maps it to an allowlisted target, a host-approved GVR +and the caller's permitted namespace, and denies by default. A host approves a GVR either from an +allowlist of its own or by delegating the decision to the upstream's authorization for that caller; +the second is valid only when the upstream is reached as the caller. The canonical scope is: ```text target, group, version, resource, [namespace], [name], [labelSelector] @@ -492,7 +515,8 @@ A **gateway** conforms iff: schema-checked or synthesized — except fields a named projection ignores. 8. It never emits, within a snapshot cycle, a state for a uid older than one already emitted. 9. It emits no SSE `id:` lines; it heartbeats; a `terminal` error is its last event and it then closes. -10. Its scope is allowlisted and server-normalized; it never accepts a raw API URL. +10. Its scope has an allowlisted target and a host-approved GVR, and is server-normalized; it never + accepts a raw API URL. 11. It satisfies the positional convergence invariant using the visible comparison in §6. > There is deliberately no rule here about rejecting a save. A gateway has no write path (§3) — the @@ -511,6 +535,8 @@ A **consumer** conforms iff: 7. It renders `binaryData` / non-UTF-8 values as opaque ("N bytes"), never as editable text. 8. It ignores heartbeats, and ignores unknown event types and unknown fields. 9. It calls `EventSource.close()` on a `terminal` error and reconnects for a fresh snapshot on a `seq` gap. +10. It treats a connection that ends after a non-terminal error as retryable, and honours `retryAfterMs` + as the least it waits before reconnecting. ### 10.1 The conformance tests that actually matter