From abeb9200f52df8a112b635c6ccddbf58298b6618 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 2 Oct 2026 13:39:09 +0000 Subject: [PATCH 1/5] ci: let releases own the adapter's core requirement and the changelogs Two release steps needed hand-fixing around 0.5.0, and both now run by themselves. The adapter's requirement on the core module was never moved by a release. Adapter changes needing unreleased core API had to pin a pseudo-version by hand, Dependabot proposed stale bumps (#36), and v0.5.0 shipped requiring a commit that lives only on a merged PR branch. Now: - gateway/kube/go.mod requires the core at its release tag, marked x-release-please-version, and Release Please rewrites it in every release PR to the version being released. - CI's go-get check builds an unreleased adapter together with the core from the same commit, and builds a release by its tag alone. It refuses a pseudo-version (the release updater would mangle it) and a missing marker. - Dependabot ignores the internal core module. Release Please also inserts each version below hand-written "Unreleased" changelog sections and never removes them, so released notes stayed labelled unreleased. File the existing notes under the versions that shipped them, and make CI refuse any section Release Please did not write. Co-Authored-By: Claude Opus 5.5 --- .github/dependabot.yml | 3 ++ .github/workflows/ci.yml | 60 ++++++++++++++++++++++---------- docs/releasing.md | 22 +++++++----- gateway/CHANGELOG.md | 50 +++++++++++++------------- gateway/kube/CHANGELOG.md | 36 ++++++++++--------- gateway/kube/go.mod | 8 ++++- gateway/kube/go.sum | 4 +-- packages/krm-stream/CHANGELOG.md | 19 +++++----- release-please-config.json | 8 ++++- 9 files changed, 127 insertions(+), 83 deletions(-) diff --git a/.github/dependabot.yml b/.github/dependabot.yml index a80e96c..e2286f1 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -38,6 +38,9 @@ updates: - dependency-name: k8s.io/api - dependency-name: k8s.io/apimachinery - dependency-name: k8s.io/client-go + # The adapter's requirement on its own core module is Release Please's, not Dependabot's: + # every release PR moves it to the version being released (see gateway/kube/go.mod). + - dependency-name: github.com/ConfigButler/krm-stream/gateway commit-message: prefix: "chore(deps)" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 28e4057..bbc85e6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -63,6 +63,17 @@ jobs: - run: hadolint .devcontainer/Dockerfile + # Release Please writes every changelog from the conventional commits it releases. It inserts + # each version below whatever heading it finds and never removes a hand-written section, so an + # "Unreleased" section ships labelled unreleased forever. Detail belongs in the commit or PR. + - name: changelogs are Release Please's + run: | + set -euo pipefail + if git ls-files -z '*CHANGELOG.md' | xargs -0 grep -nE '^## ' | grep -vE ':## (\[[0-9]+\.[0-9]+\.[0-9]+\]\(|[0-9]+\.[0-9]+\.[0-9]+ \()'; then + echo "::error::a CHANGELOG.md has a hand-written section. Release Please generates them from commit messages; put the detail in the commit or PR description instead." + exit 1 + fi + - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 with: node-version: "24" @@ -216,32 +227,42 @@ jobs: run: | set -euo pipefail - # gateway/kube requires the core module by version. On an ordinary commit that version is - # one that has already been released, so the resolution below is exactly what an adopter - # gets. On a RELEASE commit it is the version being released — whose tag does not exist - # 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)" + # gateway/kube requires the core module at the release it ships with, and Release Please + # rewrites that line in every release PR (x-release-please-version). Nobody edits it by + # hand, so it must keep its marker, and it must never be a pseudo-version: the release + # updater would mangle one (v0.6.0-e723903f0c9e), and a released adapter must not depend + # on a commit that may live only on a merged branch. + line="$(grep -E '^[[:space:]]*github.com/ConfigButler/krm-stream/gateway v' gateway/kube/go.mod | head -1)" + core="$(printf '%s' "$line" | sed -n 's|^[[:space:]]*github.com/ConfigButler/krm-stream/gateway v\([^ ]*\).*|\1|p')" if [ -z "$core" ]; then - echo "::error::gateway/kube/go.mod does not require the core module at all" && exit 1 + echo "::error file=gateway/kube/go.mod::gateway/kube/go.mod does not require the core module at all" && exit 1 + fi + if ! printf '%s' "$line" | grep -q '// x-release-please-version'; then + echo "::error file=gateway/kube/go.mod::the core requirement lost its x-release-please-version marker, so releases would stop updating it" && exit 1 + fi + if printf '%s' "$core" | grep -Eq '[-.][0-9]{14}-[0-9a-f]{12}$'; then + echo "::error file=gateway/kube/go.mod::the core requirement is a pseudo-version (v${core}). Release Please owns this line; restore the released version. CI already builds an unreleased adapter with the core from the same commit." + 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 + # On a release PR the line names the version being released, whose tag does not exist until + # the PR merges. Skip that case in ordinary CI; release CI requires the tags to exist. + 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 revision="${{ inputs.ref || github.event.pull_request.head.sha || github.sha }}" + # A release is fetched by its tag ALONE: the adapter must resolve its own core. An ordinary + # commit may use core API that is not released yet, so it is fetched the way a consumer of + # an unreleased commit must fetch it: both modules at that same commit. + modules=("github.com/ConfigButler/krm-stream/gateway/kube@${revision}" "github.com/ConfigButler/krm-stream/gateway@${revision}") if [ -n "$RELEASE_VERSION" ]; then revision="v${RELEASE_VERSION}" + modules=("github.com/ConfigButler/krm-stream/gateway/kube@${revision}") fi tmp="$(mktemp -d)" cd "$tmp" @@ -261,8 +282,9 @@ jobs: func main() {} GO - go get "github.com/ConfigButler/krm-stream/gateway/kube@${revision}" + go get "${modules[@]}" go build ./... + go list -m github.com/ConfigButler/krm-stream/gateway github.com/ConfigButler/krm-stream/gateway/kube echo "a stranger can go get and build the Kubernetes adapter at ${revision}." - name: warm the released module versions in the Go proxy diff --git a/docs/releasing.md b/docs/releasing.md index 573465a..c4a1619 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -2,6 +2,12 @@ Releases are generated from conventional commits on `main`. Release Please opens a release pull request with version bumps and changelog entries; merging that pull request creates the release. +Nothing in it needs editing by hand. + +The changelogs are written entirely from commit subjects and `BREAKING CHANGE:` footers. Do not add +sections to a `CHANGELOG.md` yourself: Release Please never removes them, so an "Unreleased" section +would ship labelled unreleased. CI refuses one. Put the detail in the commit body or the PR +description, which the changelog links to. | Commit | Release effect | |---|---| @@ -24,19 +30,17 @@ 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: +`gateway/kube/go.mod` requires the core module at the release it ships with. Every release pull +request rewrites that line to the version being released (its `x-release-please-version` marker), +so a released adapter always resolves at least its own core. Do not edit it, and do not pin a +pseudo-version when an adapter change needs unreleased core API: `go.work` wires the two modules +together in this checkout, and CI's `a stranger can go get this` job builds an unreleased adapter +with the core from the same commit. Someone using an unreleased commit does the same: ```sh -cd gateway/kube -GOWORK=off GOPROXY=direct go get github.com/ConfigButler/krm-stream/gateway@ -GOWORK=off go mod tidy +go get github.com/ConfigButler/krm-stream/gateway/kube@ github.com/ConfigButler/krm-stream/gateway@ ``` -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 d75a7fb..b2ea9f4 100644 --- a/gateway/CHANGELOG.md +++ b/gateway/CHANGELOG.md @@ -1,18 +1,18 @@ # Changelog -## Unreleased +## [0.5.0](https://github.com/ConfigButler/krm-stream/compare/gateway/v0.4.0...gateway/v0.5.0) (2026-10-02) + + +### ⚠ BREAKING CHANGES + +* Principal errors and kube.SubjectAccessReviewAuthorizer's unmapped callers are UNAUTHENTICATED instead of FORBIDDEN; unexpected error text no longer reaches the browser; non-terminal errors other than RESYNC_REQUIRED end the connection. + +### Features + +* map upstream errors, keep error text off the wire, and let the client own retries ([#40](https://github.com/ConfigButler/krm-stream/issues/40)) ([891e93e](https://github.com/ConfigButler/krm-stream/commit/891e93eed566b76a20ec651b139b30f11691afdc)) + +### Details -- Add opt-in `Options.WriteTimeout` / `Gateway.WriteTimeout` for each HTTP write plus flush; - zero installs no deadline. Header, heartbeat and flush failures now terminate the HTTP stream. -- Add `CheckHTTPStreaming` for mounted-writer tests. It clears the existing write deadline and - writes no response. Unsupported bounded transports abort and report `http_transport_rejected`; - they do not send a terminal SSE frame and clients may reconnect. -- Add `stream_closed`, `shared_subscription_opened` and `shared_subscription_closed` observations. - They describe logical streams and active attachments, not physical API watches. -- **Breaking:** remove `WriteSSEHeaders`; use `Gateway.ServeStream` or `ServeStreamProjection` to - 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. @@ -29,18 +29,6 @@ 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.5.0](https://github.com/ConfigButler/krm-stream/compare/gateway/v0.4.0...gateway/v0.5.0) (2026-10-02) - - -### ⚠ BREAKING CHANGES - -* Principal errors and kube.SubjectAccessReviewAuthorizer's unmapped callers are UNAUTHENTICATED instead of FORBIDDEN; unexpected error text no longer reaches the browser; non-terminal errors other than RESYNC_REQUIRED end the connection. - -### Features - -* map upstream errors, keep error text off the wire, and let the client own retries ([#40](https://github.com/ConfigButler/krm-stream/issues/40)) ([891e93e](https://github.com/ConfigButler/krm-stream/commit/891e93eed566b76a20ec651b139b30f11691afdc)) - ## [0.4.0](https://github.com/ConfigButler/krm-stream/compare/gateway/v0.3.0...gateway/v0.4.0) (2026-09-11) @@ -64,6 +52,20 @@ * trim obsolete history and align current integration guidance ([#34](https://github.com/ConfigButler/krm-stream/issues/34)) ([e450334](https://github.com/ConfigButler/krm-stream/commit/e450334ac19c3b39cd546e5a0e6947865afd5e3e)) +### Details + +- Add opt-in `Options.WriteTimeout` / `Gateway.WriteTimeout` for each HTTP write plus flush; + zero installs no deadline. Header, heartbeat and flush failures now terminate the HTTP stream. +- Add `CheckHTTPStreaming` for mounted-writer tests. It clears the existing write deadline and + writes no response. Unsupported bounded transports abort and report `http_transport_rejected`; + they do not send a terminal SSE frame and clients may reconnect. +- Add `stream_closed`, `shared_subscription_opened` and `shared_subscription_closed` observations. + They describe logical streams and active attachments, not physical API watches. +- **Breaking:** remove `WriteSSEHeaders`; use `Gateway.ServeStream` or `ServeStreamProjection` to + 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. + ## [0.3.0](https://github.com/ConfigButler/krm-stream/compare/gateway/v0.2.1...gateway/v0.3.0) (2026-09-11) diff --git a/gateway/kube/CHANGELOG.md b/gateway/kube/CHANGELOG.md index c44837f..3b9380e 100644 --- a/gateway/kube/CHANGELOG.md +++ b/gateway/kube/CHANGELOG.md @@ -1,22 +1,5 @@ # Changelog -## Unreleased - -- Add a tested, copyable shared ConfigMap host example with local full-subject resolution, - 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.5.0](https://github.com/ConfigButler/krm-stream/compare/gateway/kube/v0.4.0...gateway/kube/v0.5.0) (2026-10-02) @@ -28,6 +11,18 @@ * map upstream errors, keep error text off the wire, and let the client own retries ([#40](https://github.com/ConfigButler/krm-stream/issues/40)) ([891e93e](https://github.com/ConfigButler/krm-stream/commit/891e93eed566b76a20ec651b139b30f11691afdc)) +### Details + +- 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) @@ -46,6 +41,13 @@ * trim obsolete history and align current integration guidance ([#34](https://github.com/ConfigButler/krm-stream/issues/34)) ([e450334](https://github.com/ConfigButler/krm-stream/commit/e450334ac19c3b39cd546e5a0e6947865afd5e3e)) +### Details + +- Add a tested, copyable shared ConfigMap host example with local full-subject resolution, + 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. + ## [0.3.0](https://github.com/ConfigButler/krm-stream/compare/gateway/kube/v0.2.1...gateway/kube/v0.3.0) (2026-09-11) diff --git a/gateway/kube/go.mod b/gateway/kube/go.mod index 0e390d7..cdd9a70 100644 --- a/gateway/kube/go.mod +++ b/gateway/kube/go.mod @@ -20,8 +20,14 @@ go 1.27.1 // So we sit on the floor and let the consumer pick the ceiling. // // Raise these only to raise the SUPPORTED MINOR, and say why in the commit. +// +// The core requirement is the opposite: it is always the release this adapter ships with, and +// Release Please rewrites it in every release PR (the marker below). Never edit it by hand, and never +// pin a pseudo-version: go.work wires the two modules together in this checkout, and CI builds an +// unreleased adapter with the core from the same commit. The adapter relies on core behaviour, not +// only API, so it must never resolve an older core than the one it was released with. require ( - github.com/ConfigButler/krm-stream/gateway v0.4.1-0.20261002130336-e723903f0c9e + github.com/ConfigButler/krm-stream/gateway v0.5.0 // x-release-please-version 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 34f5e52..4706f1f 100644 --- a/gateway/kube/go.sum +++ b/gateway/kube/go.sum @@ -1,5 +1,5 @@ -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/ConfigButler/krm-stream/gateway v0.5.0 h1:fvu6IM1AKfpjVR/oWOKcKVoUo0K0gkdwjMkZdLrBQoI= +github.com/ConfigButler/krm-stream/gateway v0.5.0/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/packages/krm-stream/CHANGELOG.md b/packages/krm-stream/CHANGELOG.md index 277ee9e..8657934 100644 --- a/packages/krm-stream/CHANGELOG.md +++ b/packages/krm-stream/CHANGELOG.md @@ -1,15 +1,5 @@ # 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.5.0](https://github.com/ConfigButler/krm-stream/compare/@configbutler/krm-stream-v0.4.0...@configbutler/krm-stream-v0.5.0) (2026-10-02) @@ -21,6 +11,15 @@ * map upstream errors, keep error text off the wire, and let the client own retries ([#40](https://github.com/ConfigButler/krm-stream/issues/40)) ([891e93e](https://github.com/ConfigButler/krm-stream/commit/891e93eed566b76a20ec651b139b30f11691afdc)) +### Details + +- 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/release-please-config.json b/release-please-config.json index 0dfd381..5ef6716 100644 --- a/release-please-config.json +++ b/release-please-config.json @@ -22,7 +22,13 @@ "gateway/kube": { "release-type": "go", "component": "gateway/kube", - "tag-separator": "/" + "tag-separator": "/", + "extra-files": [ + { + "type": "generic", + "path": "go.mod" + } + ] } }, "plugins": [ From ad999de854a50e3d71bf8d5d36cdc45b7d01e5ab Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 2 Oct 2026 18:55:42 +0000 Subject: [PATCH 2/5] feat: refuse redirects that would carry the caller's credential, and stop reopening watches that end early MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses asks 6 to 8 of krm-foyer's renewed field report. Ask 6: client-go follows redirects and puts the configured credential on the redirected request, so a backend holding the caller's token sent it wherever a redirect pointed. kube.NewBackendForConfig now builds its client with the new kube.HTTPClientFor, which refuses redirects; a refused redirect is a terminal INTERNAL (it will not go away on retry) whose cause is kube.ErrRedirectRefused. The shared-host example and the docs use the refusing client. Ask 7: the stream loop reopened at once after ErrWatchClosed or RESYNC_REQUIRED, so an upstream ending every watch before its snapshot completed, or straight after, was reopened in a tight loop on one connection. A cycle now counts as useful only once it has completed its snapshot and stayed live for a second. The first early end is still recovered in band (spec §5); a second in a row ends the connection with a non-terminal UPSTREAM_UNAVAILABLE. A StreamError now decides before ErrWatchClosed, so an UPSTREAM_UNAVAILABLE caused by a closed watch is no longer read as a routine close through Unwrap. Ask 8: the operations guide says client-go retries a 429 or 5xx carrying Retry-After up to ten times before the gateway sees it. Those retries stay on: the API server sets the pace, the browser holds a heartbeating connection in "syncing" meanwhile, and turning them off would trade one paced wait for a reconnect and fresh snapshot from every browser. Also record krm-foyer's renewed report, with its links made absolute. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- docs/adopting.md | 13 ++- docs/auth.md | 6 +- docs/field-reports/first-fieldreport.md | 52 +++++++++ docs/operations.md | 18 +++ gateway/README.md | 2 +- gateway/auth_test.go | 17 +++ gateway/kube/backend.go | 38 +++++- gateway/kube/examples/sharedstream/handler.go | 10 +- gateway/kube/redirect_test.go | 55 +++++++++ gateway/stream.go | 82 ++++++++++--- gateway/stream_test.go | 109 ++++++++++++++++++ spec/v1.md | 4 +- 13 files changed, 384 insertions(+), 24 deletions(-) create mode 100644 gateway/kube/redirect_test.go diff --git a/README.md b/README.md index 13d02e7..6fc6a24 100644 --- a/README.md +++ b/README.md @@ -151,7 +151,7 @@ mux.Handle("/resource-stream/v1", gateway.Handler(gateway.Options{ Principal: func(r *http.Request) (gateway.Principal, error) { return userFromSession(r) }, Authorizer: authorizeScope, Clients: func(_ context.Context, _ string, p gateway.Principal) (gateway.Backend, error) { - return kube.NewBackend(dynamicClientFor(p.(*User))), nil + return kube.NewBackendForConfig(restConfigFor(p.(*User))) }, Scopes: gateway.ScopePolicy{ Targets: []string{"production"}, diff --git a/docs/adopting.md b/docs/adopting.md index bdf740c..14c95de 100644 --- a/docs/adopting.md +++ b/docs/adopting.md @@ -75,7 +75,10 @@ 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 errors, which reach `Options.Diagnostics` and never the browser; -`kube.NewBackend(dynamicClient)` also works when a dynamic client already exists. +`kube.NewBackend(dynamicClient)` also works when a dynamic client already exists. Build that client on +`kube.HTTPClientFor(cfg)` (`dynamic.NewForConfigAndClient(cfg, httpClient)`): client-go follows +redirects and carries the configured credential along, so a client holding a user's token must refuse +them, as `NewBackendForConfig`'s does. ## 2. Mount the same-origin cookie endpoint @@ -83,11 +86,13 @@ requests. Use `kube.NewBackendForConfig(cfg)` to include the endpoint in upstrea import ( "net/http" + "k8s.io/client-go/rest" + "github.com/ConfigButler/krm-stream/gateway" "github.com/ConfigButler/krm-stream/gateway/kube" ) -func mount(mux *http.ServeMux, dynamicClientFor func(*User) dynamic.Interface) { +func mount(mux *http.ServeMux, restConfigFor func(*User) *rest.Config) { mux.Handle("/resource-stream/v1", gateway.Handler(gateway.Options{ // Your session cookie -> your application user. krm-stream never sees a token. Principal: func(r *http.Request) (gateway.Principal, error) { @@ -101,9 +106,9 @@ func mount(mux *http.ServeMux, dynamicClientFor func(*User) dynamic.Interface) { } return nil }), - // Build a dynamic client acting as this user. Kubernetes RBAC remains the boundary. + // Build a backend acting as this user. Kubernetes RBAC remains the boundary. Clients: func(_ context.Context, _ string, p gateway.Principal) (gateway.Backend, error) { - return kube.NewBackend(dynamicClientFor(p.(*User))), nil + return kube.NewBackendForConfig(restConfigFor(p.(*User))) }, Scopes: gateway.ScopePolicy{ Targets: []string{"production"}, diff --git a/docs/auth.md b/docs/auth.md index 0bfc2df..c4e5297 100644 --- a/docs/auth.md +++ b/docs/auth.md @@ -47,7 +47,7 @@ gateway.Handler(gateway.Options{ Principal: sessionUser, Authorizer: authorizeScope, Clients: func(_ context.Context, target string, p gateway.Principal) (gateway.Backend, error) { - return kube.NewBackend(dynamicClientFor(target, p.(*User))), nil + return kube.NewBackendForConfig(restConfigFor(target, p.(*User))) }, Scopes: scopePolicy, }) @@ -58,6 +58,10 @@ gateway.Handler(gateway.Options{ `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. + `kube.NewBackendForConfig` builds a client that refuses redirects, since client-go would + otherwise carry the caller's token to wherever a redirect points. A host that builds its own + dynamic client for `kube.NewBackend` must build it on `kube.HTTPClientFor(cfg)` for the same + reason. - `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. diff --git a/docs/field-reports/first-fieldreport.md b/docs/field-reports/first-fieldreport.md index 3266fe1..ef494f5 100644 --- a/docs/field-reports/first-fieldreport.md +++ b/docs/field-reports/first-fieldreport.md @@ -7,6 +7,18 @@ 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. +## Status (2026-10-02): asks 1 to 5 are in krm-stream 0.5.0 + +krm-stream 0.5.0 took all five asks below (krm-stream #40), and krm-foyer moved to it: +the gateway maps the API server's answers, keeps their text off the wire, sends it to a +`Diagnostics` hook, leaves retries to the browser's client with the server's hint, admits +any resource with `ScopePolicy.AnyResource`, and lets `Principal` choose its refusal. +krm-foyer's own mapping, status capture and backoff are gone. + +Reviewing krm-foyer's integration found three more, below the original asks as +[asks 6 to 8](#ask-6-refuse-redirects-in-the-kubernetes-backend). krm-foyer works around +each meanwhile. + ## The asks at a glance | # | Ask | Priority | What krm-foyer does meanwhile | @@ -208,3 +220,43 @@ Smaller, and related: on a non-200 answer the connector reports only `stream: HT 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. + +## Ask 6: refuse redirects in the Kubernetes backend + +**This one is about credentials.** client-go follows redirects, and its transport adds the +bearer token to the redirected request too. A backend built with +`kube.NewBackendForConfig(cfg)` from a `rest.Config` holding the caller's token therefore +sends that token wherever a redirect points: another host, or plain `http`. Pinning the +host and leaving `Proxy` unset does not prevent it. krm-foyer reproduced it against a +test API server that redirects to a recording server, which received the token. An +aggregated API's answer passes back through the API server, so a redirect is not only a +hostile-server case. + +krm-foyer builds its client with `rest.HTTPClientFor`, sets `CheckRedirect` to refuse, +and ends the stream with a terminal error. We suggest `NewBackendForConfig` do the same, +and that the gateway map a refused redirect to a terminal error, since a redirect will +not go away on retry. The docs that show `kube.NewBackend(dynamicClientFor(user))` +should say a host-built client must refuse redirects too. + +## Ask 7: treat a watch that ends before it was of use as a failure + +The stream loop reopens at once, on the same connection, after `ErrWatchClosed`, and +after a `RESYNC_REQUIRED`. That is right for the API server's routine timeout, which +comes after half an hour. But an upstream that ends every watch before its snapshot is +complete, or answers 410 right after it, gets a tight loop: krm-foyer measured about 100 +reopenings a second against a test server that does so, on one browser connection. + +krm-foyer's backend now answers such an end with a non-terminal `UPSTREAM_UNAVAILABLE`, +so the connection closes and the client waits: an end before the `synced` boundary, or +within a second of it. We suggest the gateway do this itself, for every backend. One +detail cost us a bug: `StreamError.Unwrap` returns `Cause`, so an `UPSTREAM_UNAVAILABLE` +whose cause is `ErrWatchClosed` is still read by the loop as a routine close. + +## Ask 8: say that client-go retries a 429 with Retry-After itself + +client-go asks again on its own, up to ten times, when a 429 or 5xx answer carries a +`Retry-After` header, before the backend sees the error. The API server sets the pace, so +this is not a tight loop, but the browser hears nothing meanwhile, and the +`retryAfterMs` mapping of 0.5.0 only applies once client-go has given up. We suggest +saying so in the operations guide, and considering a client with those retries off, +so the browser's client, which shows the wait, owns them as 0.5.0 intends. diff --git a/docs/operations.md b/docs/operations.md index aa8e530..e8b83f4 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -29,6 +29,24 @@ as `INTERNAL` "internal error". Decide what to log there, and redact it. | 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 | +## When the API server pushes back + +client-go retries on its own before the gateway sees an error. A 429 from Priority and Fairness, or +any 5xx, that carries a `Retry-After` header is retried up to ten times at the pace that header sets. +Only when client-go gives up does the gateway map the answer, to `UPSTREAM_UNAVAILABLE` with +`retryAfterMs`, and close the connection for the browser's client to retry. + +While client-go waits, the browser holds an open connection with heartbeats and no snapshot yet: the +managed connector reports `syncing`, not an error. That is deliberate. The API server sets the pace, +so it is not a tight loop. Turning these retries off would mean stripping `Retry-After` in a wrapping +transport, and would turn one paced wait into a reconnect and a fresh snapshot from every browser. +A rising `syncing` time, without errors, is the signal to look at API-server load. + +A watch that ends before it is of use (before its snapshot completes, or within a second of it) is +recovered once on the same connection. A second such end in a row closes the connection with +`UPSTREAM_UNAVAILABLE`, so an upstream that ends every watch early meets the client's backoff and not +a reconnect loop. It shows as `retryable_error`. + ## Runtime controls | Control | Default | Use | diff --git a/gateway/README.md b/gateway/README.md index d24189d..49c730a 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -25,7 +25,7 @@ handler := gateway.Handler(gateway.Options{ Principal: principalFromSession, Authorizer: authorizeScope, Clients: func(_ context.Context, target string, p gateway.Principal) (gateway.Backend, error) { - return kube.NewBackend(dynamicClientFor(target, p)), nil + return kube.NewBackendForConfig(restConfigFor(target, p)) }, Scopes: gateway.ScopePolicy{ Targets: []string{"production"}, diff --git a/gateway/auth_test.go b/gateway/auth_test.go index 3305dfc..aaa5dcd 100644 --- a/gateway/auth_test.go +++ b/gateway/auth_test.go @@ -3,8 +3,10 @@ package gateway import ( "context" "strings" + "sync" "sync/atomic" "testing" + "time" ) // Authorization and credentials, across the LIFE of a stream — not merely at its start. @@ -108,6 +110,9 @@ func TestTheClientIsResolvedOnEveryCycle(t *testing.T) { } return backend, nil }, + // Each cycle stands in for a long-lived one: without this, instant cycles are an upstream + // ending every watch early, which the gateway stops reopening after the second. + now: steppingClock(2 * time.Second), } _ = g.Stream(t.Context(), "alice", Scope{Version: "v1", Resource: "configmaps"}, sink) @@ -161,3 +166,15 @@ func TestDenialOpensNoWatchAtAll(t *testing.T) { t.Errorf("want a single terminal FORBIDDEN, got: %s", sink.codes()) } } + +// steppingClock returns a clock that advances by step on every reading. +func steppingClock(step time.Duration) func() time.Time { + var mu sync.Mutex + now := time.Unix(0, 0) + return func() time.Time { + mu.Lock() + defer mu.Unlock() + now = now.Add(step) + return now + } +} diff --git a/gateway/kube/backend.go b/gateway/kube/backend.go index 4830128..ede31b2 100644 --- a/gateway/kube/backend.go +++ b/gateway/kube/backend.go @@ -30,6 +30,7 @@ import ( "errors" "fmt" "net" + "net/http" "strings" "sync" "time" @@ -97,8 +98,14 @@ func NewBackend(client dynamic.Interface) *Backend { // wrongly — two /clusters/ segments, say — the API server answers "the server could not find the // requested resource", which is indistinguishable from "that CRD is not installed" until the error // tells you which URL it dialed. Now it does. +// +// Its client refuses redirects: see HTTPClientFor. func NewBackendForConfig(cfg *rest.Config) (*Backend, error) { - client, err := dynamic.NewForConfig(cfg) + httpClient, err := HTTPClientFor(cfg) + if err != nil { + return nil, fmt.Errorf("krm-stream/kube: HTTP client for %s: %w", cfg.Host, err) + } + client, err := dynamic.NewForConfigAndClient(cfg, httpClient) if err != nil { return nil, fmt.Errorf("krm-stream/kube: dynamic client for %s: %w", cfg.Host, err) } @@ -107,6 +114,29 @@ func NewBackendForConfig(cfg *rest.Config) (*Backend, error) { return b, nil } +// ErrRedirectRefused is the cause of a request that the API server answered with a redirect. +var ErrRedirectRefused = errors.New("krm-stream/kube: the API server answered with a redirect, which is refused") + +// HTTPClientFor is rest.HTTPClientFor with redirects refused. +// +// client-go follows redirects, and its transport puts the configured credential (a bearer token, a +// client certificate's TLS session) on the redirected request too. A client holding the caller's +// token would hand it to wherever a redirect points: another host, or plain http. Pinning the host +// and leaving Proxy unset does not prevent it. An aggregated API's answer passes back through the +// API server, so this is not only a hostile-server case. +// +// NewBackendForConfig uses it. A host that builds its own dynamic client for NewBackend should build +// it on this client: dynamic.NewForConfigAndClient(cfg, httpClient). A refused redirect reaches the +// browser as a terminal INTERNAL, since it will not go away on retry; ErrRedirectRefused is its cause. +func HTTPClientFor(cfg *rest.Config) (*http.Client, error) { + client, err := rest.HTTPClientFor(cfg) + if err != nil { + return nil, err + } + client.CheckRedirect = func(*http.Request, []*http.Request) error { return ErrRedirectRefused } + return client, nil +} + // upstream describes what a failing call was actually reaching for: the server, and the resource path // built from the scope. Both halves are what a "could not find the requested resource" needs before // it can be diagnosed rather than guessed at. @@ -393,6 +423,12 @@ func watchError(obj runtime.Object) error { // 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 { + // First: a refused redirect arrives as a *url.Error, which would otherwise read as a transport + // failure and be retried, and it will be refused again every time. + if errors.Is(err, ErrRedirectRefused) { + return withCause(&gateway.StreamError{Code: gateway.CodeInternal, Terminal: true, + Message: "the API server answered with a redirect, which this gateway refuses to follow"}, err) + } if se := classifyStatus(err); se != nil { return se } diff --git a/gateway/kube/examples/sharedstream/handler.go b/gateway/kube/examples/sharedstream/handler.go index 103d2c8..7317728 100644 --- a/gateway/kube/examples/sharedstream/handler.go +++ b/gateway/kube/examples/sharedstream/handler.go @@ -45,11 +45,17 @@ func Handler(cluster *rest.Config, namespace, name string, sessionFor func(*http if !rest.IsConfigTransportTLS(*cluster) || cluster.Insecure { return nil, fmt.Errorf("sharedstream: cluster must use HTTPS with certificate verification") } - service, err := kubernetes.NewForConfig(cluster) + // One client for both, and it refuses redirects: client-go would otherwise carry the service + // token to wherever a redirect points. + httpClient, err := kube.HTTPClientFor(cluster) if err != nil { return nil, err } - data, err := dynamic.NewForConfig(cluster) + service, err := kubernetes.NewForConfigAndClient(cluster, httpClient) + if err != nil { + return nil, err + } + data, err := dynamic.NewForConfigAndClient(cluster, httpClient) if err != nil { return nil, err } diff --git a/gateway/kube/redirect_test.go b/gateway/kube/redirect_test.go new file mode 100644 index 0000000..27d6293 --- /dev/null +++ b/gateway/kube/redirect_test.go @@ -0,0 +1,55 @@ +package kube + +import ( + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "k8s.io/client-go/rest" + + "github.com/ConfigButler/krm-stream/gateway" +) + +// client-go follows redirects, and its transport puts the bearer token on the redirected request +// too. A backend holding the caller's token would hand it to wherever a redirect points: another +// host, or plain http. An aggregated API's answer passes back through the API server, so this is +// not only a hostile-server case. The backend refuses redirects, and the refusal is terminal: a +// redirect does not go away on retry. +func TestABackendNeverFollowsARedirectWithTheCallersToken(t *testing.T) { + const token = "caller-token-that-must-not-travel" + var reached atomic.Int32 + recorder := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached.Add(1) + if strings.Contains(r.Header.Get("Authorization"), token) { + t.Errorf("the caller's token reached the redirect target: %s %s", r.Method, r.URL) + } + http.Error(w, "recorded", http.StatusTeapot) + })) + defer recorder.Close() + apiServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, recorder.URL+"/elsewhere", http.StatusTemporaryRedirect) + })) + defer apiServer.Close() + + b, err := NewBackendForConfig(&rest.Config{Host: apiServer.URL, BearerToken: token}) + if err != nil { + t.Fatal(err) + } + _, err = b.Watch(t.Context(), scope) + if n := reached.Load(); n != 0 { + t.Fatalf("the redirect was followed %d times", n) + } + var se *gateway.StreamError + if !errors.As(err, &se) || se.Code != gateway.CodeInternal || !se.Terminal { + t.Fatalf("err = %v, want a terminal INTERNAL: a redirect does not go away on retry", err) + } + if strings.Contains(se.Message, recorder.URL) || strings.Contains(se.Message, apiServer.URL) { + t.Errorf("message %q names an address", se.Message) + } + if !errors.Is(err, ErrRedirectRefused) { + t.Errorf("err = %v, want it to wrap ErrRedirectRefused for the host's diagnostics", err) + } +} diff --git a/gateway/stream.go b/gateway/stream.go index 0857310..37c50b5 100644 --- a/gateway/stream.go +++ b/gateway/stream.go @@ -103,6 +103,9 @@ type Gateway struct { // ReauthorizationTimeout bounds a timed check. Zero defaults to 10 seconds. // Authorizers and projection policies must honor context cancellation. ReauthorizationTimeout time.Duration + // now is the clock that decides whether a cycle lasted long enough to count. Nil uses time.Now. + now func() time.Time + // Ordering is how far the upstream's resourceVersions may be trusted. The zero value is // OrderingStrict: this library targets Kubernetes 1.35+, where orderability is a conformance // requirement, and it says so rather than degrading quietly on every cluster to accommodate one. @@ -132,6 +135,12 @@ func (g *Gateway) StreamProjection(ctx context.Context, principal Principal, sco policy = StaticProjection(g.Projection) } + // quickEnds counts consecutive cycles that ended before they were of use. One is recovered in + // band, as continuity loss always is (§5). A second in a row is an upstream that ends every + // watch early — before its snapshot completes, or straight after — and reopening it at once + // on this connection would be a tight loop. That ends the connection instead, and the client + // waits on its own budget. + quickEnds := 0 for cycles := 0; ; cycles++ { if cycles > 0 { g.observe(Observation{Kind: ObservationConsumerResync, Scope: scope}) @@ -171,32 +180,79 @@ func (g *Gateway) StreamProjection(ctx context.Context, principal Principal, sco return g.emitError(ctx, sink, principal, scope, projection, err) } - err = g.authorizedCycle(ctx, principal, scope, requested, policy, backend, projection, revisions, sink) + tracker := &syncTracker{sink: sink, now: g.clock} + err = g.authorizedCycle(ctx, principal, scope, requested, policy, backend, projection, revisions, tracker) switch { case err == nil: // A cycle only ends by error or cancellation; nil would be a bug in the loop below. return nil case ctx.Err() != nil: return ctx.Err() - case errors.Is(err, ErrWatchClosed): - // A routine watch timeout. Reopen — with a full snapshot, because between the close and - // the reopen we can promise nothing, and the protocol's whole value is that what you - // end up holding is right. - 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. + // Two endings are recovered with a new cycle on this connection: a routine watch timeout + // (ErrWatchClosed), because between the close and the reopen we can promise nothing; and lost + // continuity (a 410, a partial object, an overflowing shared queue), announced as + // RESYNC_REQUIRED. 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. + // + // A StreamError decides before ErrWatchClosed does. StreamError.Unwrap exposes its Cause, so + // an UPSTREAM_UNAVAILABLE caused by a closed watch also satisfies errors.Is(err, + // ErrWatchClosed), and must not be mistaken for a routine timeout. var se *StreamError - if errors.As(err, &se) && se != nil && !se.Terminal && se.Code == CodeResyncRequired { + hasStreamError := errors.As(err, &se) && se != nil + recoverable := hasStreamError && !se.Terminal && se.Code == CodeResyncRequired || + !hasStreamError && errors.Is(err, ErrWatchClosed) + if !recoverable { + return g.emitError(ctx, sink, principal, scope, projection, err) + } + if hasStreamError { g.diagnose(Diagnostic{Principal: principal, Scope: scope, Code: se.Code, Err: err}) + } + if tracker.lastedPast(minUsefulCycle) { + quickEnds = 0 continue } - return g.emitError(ctx, sink, principal, scope, projection, err) + if quickEnds++; quickEnds > 1 { + unavailable := UpstreamUnavailable("the upstream watch keeps ending before it is of use; retry later", 0) + unavailable.Cause = err + return g.emitError(ctx, sink, principal, scope, projection, unavailable) + } + } +} + +// minUsefulCycle is how long a cycle must stay live after its snapshot to count as useful. The API +// server's routine watch timeout comes after many minutes; an upstream that ends a watch within a +// second of its snapshot is failing, not timing out. +const minUsefulCycle = time.Second + +func (g *Gateway) clock() time.Time { + if g.now != nil { + return g.now() } + return time.Now() +} + +// syncTracker passes a cycle's events through and remembers when its snapshot completed. +type syncTracker struct { + sink Sink + now func() time.Time + syncedAt time.Time +} + +func (t *syncTracker) Emit(ctx context.Context, ev Event) error { + err := t.sink.Emit(ctx, ev) + if err == nil && ev.Type == EventSynced && t.syncedAt.IsZero() { + t.syncedAt = t.now() + } + return err +} + +// lastedPast reports whether the cycle completed its snapshot and then stayed live for at least d. +func (t *syncTracker) lastedPast(d time.Duration) bool { + return !t.syncedAt.IsZero() && t.now().Sub(t.syncedAt) >= d } func isBuiltinProjection(projection Projection) bool { diff --git a/gateway/stream_test.go b/gateway/stream_test.go index 3612e0a..6666e0f 100644 --- a/gateway/stream_test.go +++ b/gateway/stream_test.go @@ -727,3 +727,112 @@ func TestOnlyContinuityLossIsRecoveredInBand(t *testing.T) { type backendFunc func() (Watcher, error) func (f backendFunc) Watch(context.Context, Scope) (Watcher, error) { return f() } + +// scriptedEnds hands out one watcher per open, each ending the way the script says. +type scriptedEnds struct { + opens int + next func(open int) Watcher +} + +func (b *scriptedEnds) Watch(context.Context, Scope) (Watcher, error) { + b.opens++ + return b.next(b.opens), nil +} + +// endsWith replays events and then fails with err. +type endsWith struct { + events []WatchEvent + err error +} + +func (w *endsWith) Next(context.Context) (WatchEvent, error) { + if len(w.events) > 0 { + ev := w.events[0] + w.events = w.events[1:] + return ev, nil + } + return WatchEvent{}, w.err +} +func (*endsWith) Stop() {} + +var boundary = WatchEvent{Type: WatchBookmark, InitialEventsEnd: true} + +func streamOver(t *testing.T, b Backend, now func() time.Time) (*recordingSink, error) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + sink := &recordingSink{} + gw := &Gateway{Auth: AllowAll{}, now: now, Clients: func(context.Context, string, Principal) (Backend, error) { return b, nil }} + err := gw.Stream(ctx, nil, Scope{Version: "v1", Resource: "configmaps"}, sink) + if ctx.Err() != nil { + t.Fatalf("the stream did not end: %d events, the gateway is reopening a failing upstream in a loop", len(sink.events)) + } + return sink, err +} + +// An upstream that ends every watch before it is of use (before its snapshot completes, or with a +// 410 straight after) would be reopened at once, forever, on one connection. The first early end +// is recovered in band like any continuity loss; the second in a row ends the connection with a +// retryable error, so the client's backoff takes over. +func TestAnUpstreamThatKeepsEndingEarlyIsNotReopenedInALoop(t *testing.T) { + for name, watcher := range map[string]func() Watcher{ + "closed before its snapshot": func() Watcher { return &endsWith{err: ErrWatchClosed} }, + "410 right after its snapshot": func() Watcher { + return &endsWith{events: []WatchEvent{boundary}, err: ResyncRequired("the upstream resourceVersion expired (410 Gone)")} + }, + } { + t.Run(name, func(t *testing.T) { + b := &scriptedEnds{next: func(int) Watcher { return watcher() }} + sink, err := streamOver(t, b, nil) + 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 b.opens != 2 { + t.Errorf("upstream opens = %d, want 2: one in-band recovery, then the client's backoff", b.opens) + } + if last := sink.events[len(sink.events)-1]; last.Code != CodeUpstreamUnavailable || last.Terminal { + t.Errorf("last event = %+v", last) + } + }) + } +} + +// One early end between useful cycles is ordinary continuity loss, and resets nothing: the count is +// of consecutive early ends. +func TestAUsefulCycleResetsTheEarlyEndCount(t *testing.T) { + b := &scriptedEnds{next: func(open int) Watcher { + switch { + case open > 6: + return &endsWith{err: Forbidden("stop")} // ends the test + case open%2 == 1: + return &endsWith{err: ErrWatchClosed} // early: before its snapshot + default: + return &endsWith{events: []WatchEvent{boundary}, err: ErrWatchClosed} // useful, by the clock + } + }} + _, err := streamOver(t, b, steppingClock(2*time.Second)) + var se *StreamError + if !errors.As(err, &se) || se.Code != CodeForbidden { + t.Fatalf("Stream() = %v after %d opens, want the scripted FORBIDDEN: alternating early ends were treated as a failing upstream", err, b.opens) + } +} + +// StreamError.Unwrap exposes its Cause, so an UPSTREAM_UNAVAILABLE caused by a closed watch also +// matches errors.Is(err, ErrWatchClosed). It must still end the connection, not be reopened as a +// routine timeout. +func TestAStreamErrorCausedByAClosedWatchIsNotARoutineClose(t *testing.T) { + b := &scriptedEnds{next: func(int) Watcher { + se := UpstreamUnavailable("the upstream ended the watch early", 0) + se.Cause = ErrWatchClosed + return &endsWith{events: []WatchEvent{boundary}, err: se} + }} + _, err := streamOver(t, b, steppingClock(2*time.Second)) + var se *StreamError + if !errors.As(err, &se) || se.Code != CodeUpstreamUnavailable { + t.Fatalf("Stream() = %v", err) + } + if b.opens != 1 { + t.Errorf("upstream opens = %d, want 1: the error was read as a routine close and reopened", b.opens) + } +} diff --git a/spec/v1.md b/spec/v1.md index af4f573..f797dff 100644 --- a/spec/v1.md +++ b/spec/v1.md @@ -257,7 +257,9 @@ stable client behavior. v1 codes: **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). + follows it immediately (§5). A gateway whose upstream keeps ending watches before they are of use + (before the snapshot completes, or straight after it) MUST NOT keep reopening them on the same + connection: after repeated such ends it sends `UPSTREAM_UNAVAILABLE` and closes, as below. - 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 From ec3dc6e73f8b4d150995dce80ce57db5da3df810 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 2 Oct 2026 19:03:37 +0000 Subject: [PATCH 3/5] fix: refuse redirects for participant clients, keep http.DefaultClient intact, and back off shared watches that end early Review follow-ups for asks 6 to 8. - The shared-host example's participant SelfSubjectReview client carried the participant's token through a client that followed redirects. It is now built on kube.HTTPClientFor too. - rest.HTTPClientFor returns http.DefaultClient for a configuration needing no special transport, so installing the redirect policy on it changed every HTTP caller in the process. HTTPClientFor now copies the client first. - The stream loop bounded each connection to two early cycles, but a SharedBackend dropped an early-ending upstream watch without any backoff, so a cohort of reconnecting subscribers still reopened it once each. The shared scope now applies the same rule to its upstream: a watch is of use once it has completed its snapshot and stayed live for a second; one early end is tolerated, a second in a row backs off. Only a useful watch's end resets the backoff, which also removes the delayed-recovery race structurally. - The operations guide no longer claims that turning client-go's retries off requires stripping Retry-After; it states why the default is kept. Co-Authored-By: Claude Opus 5.5 --- docs/operations.md | 9 +- gateway/kube/backend.go | 7 +- gateway/kube/examples/sharedstream/handler.go | 7 +- .../examples/sharedstream/handler_test.go | 48 +++++++ gateway/kube/redirect_test.go | 24 ++++ gateway/shared.go | 134 +++++++++++++----- gateway/shared_test.go | 116 ++++++++++++--- spec/v1.md | 3 +- 8 files changed, 285 insertions(+), 63 deletions(-) diff --git a/docs/operations.md b/docs/operations.md index e8b83f4..2ac3870 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -37,10 +37,11 @@ Only when client-go gives up does the gateway map the answer, to `UPSTREAM_UNAVA `retryAfterMs`, and close the connection for the browser's client to retry. While client-go waits, the browser holds an open connection with heartbeats and no snapshot yet: the -managed connector reports `syncing`, not an error. That is deliberate. The API server sets the pace, -so it is not a tight loop. Turning these retries off would mean stripping `Retry-After` in a wrapping -transport, and would turn one paced wait into a reconnect and a fresh snapshot from every browser. -A rising `syncing` time, without errors, is the signal to look at API-server load. +managed connector reports `syncing`, not an error. krm-stream keeps client-go's default on purpose. +The API server sets the pace, so it is not a tight loop, and the wait happens once per stream at the +API server's request, where a reconnect would cost every browser a new connection and a fresh +snapshot. The cost is that the browser is not told why it is waiting. A rising `syncing` time, +without errors, is the signal to look at API-server load. A watch that ends before it is of use (before its snapshot completes, or within a second of it) is recovered once on the same connection. A second such end in a row closes the connection with diff --git a/gateway/kube/backend.go b/gateway/kube/backend.go index ede31b2..ca0e8a4 100644 --- a/gateway/kube/backend.go +++ b/gateway/kube/backend.go @@ -129,12 +129,15 @@ var ErrRedirectRefused = errors.New("krm-stream/kube: the API server answered wi // it on this client: dynamic.NewForConfigAndClient(cfg, httpClient). A refused redirect reaches the // browser as a terminal INTERNAL, since it will not go away on retry; ErrRedirectRefused is its cause. func HTTPClientFor(cfg *rest.Config) (*http.Client, error) { - client, err := rest.HTTPClientFor(cfg) + shared, err := rest.HTTPClientFor(cfg) if err != nil { return nil, err } + // A copy: for a configuration needing no special transport, client-go returns http.DefaultClient + // itself, and setting a policy on that would change every HTTP caller in the process. + client := *shared client.CheckRedirect = func(*http.Request, []*http.Request) error { return ErrRedirectRefused } - return client, nil + return &client, nil } // upstream describes what a failing call was actually reaching for: the server, and the resource path diff --git a/gateway/kube/examples/sharedstream/handler.go b/gateway/kube/examples/sharedstream/handler.go index 7317728..0c4dfe9 100644 --- a/gateway/kube/examples/sharedstream/handler.go +++ b/gateway/kube/examples/sharedstream/handler.go @@ -79,7 +79,12 @@ func Handler(cluster *rest.Config, namespace, name string, sessionFor func(*http } cfg := rest.CopyConfig(participantConfig) cfg.BearerToken = state.session.Token - client, err := kubernetes.NewForConfig(cfg) + // The participant's token, so redirects are refused here too. + httpClient, err := kube.HTTPClientFor(cfg) + if err != nil { + return nil, err + } + client, err := kubernetes.NewForConfigAndClient(cfg, httpClient) if err != nil { return nil, err } diff --git a/gateway/kube/examples/sharedstream/handler_test.go b/gateway/kube/examples/sharedstream/handler_test.go index b34ad4d..2b30924 100644 --- a/gateway/kube/examples/sharedstream/handler_test.go +++ b/gateway/kube/examples/sharedstream/handler_test.go @@ -11,6 +11,7 @@ import ( "reflect" "strings" "sync" + "sync/atomic" "testing" "time" @@ -222,3 +223,50 @@ func TestCounterMappingConcurrentAndUnknown(t *testing.T) { t.Fatal("counter mapping is unbalanced") } } + +// The participant's SelfSubjectReview client holds the participant's token, so it must refuse +// redirects as the service clients do: client-go would carry the token to wherever one points. +func TestParticipantIdentityResolutionRefusesRedirects(t *testing.T) { + var stolen atomic.Int32 + api := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "selfsubjectreviews"): + // Same server, so the target's certificate is trusted: only the redirect policy stands + // between the token and the recording path. + http.Redirect(w, r, "/stolen", http.StatusTemporaryRedirect) + case r.URL.Path == "/stolen": + stolen.Add(1) + http.Error(w, "recorded", http.StatusTeapot) + default: + http.NotFound(w, r) + } + })) + defer api.Close() + cluster := &rest.Config{ + Host: api.URL, + BearerToken: "service", + ContentConfig: rest.ContentConfig{ContentType: "application/json"}, + TLSClientConfig: rest.TLSClientConfig{CAData: pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: api.Certificate().Raw})}, + } + session := func(*http.Request) (Session, error) { + return Session{Token: "participant", SessionExpiry: time.Now().Add(time.Hour), TokenExpiry: time.Now().Add(time.Hour)}, nil + } + h, err := Handler(cluster, "app", "coffee", session, &Counters{}) + if err != nil { + t.Fatal(err) + } + host := httptest.NewServer(h) + defer host.Close() + res, err := host.Client().Get(host.URL + "/?version=v1&resource=configmaps&namespace=app&name=coffee") + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(res.Body) + _ = res.Body.Close() + if n := stolen.Load(); n != 0 { + t.Fatalf("the participant's SelfSubjectReview followed a redirect %d times", n) + } + if !strings.Contains(string(body), `"type":"error"`) || strings.Contains(string(body), "synced") { + t.Fatalf("an unresolved participant was streamed: %s", body) + } +} diff --git a/gateway/kube/redirect_test.go b/gateway/kube/redirect_test.go index 27d6293..89ea4dc 100644 --- a/gateway/kube/redirect_test.go +++ b/gateway/kube/redirect_test.go @@ -53,3 +53,27 @@ func TestABackendNeverFollowsARedirectWithTheCallersToken(t *testing.T) { t.Errorf("err = %v, want it to wrap ErrRedirectRefused for the host's diagnostics", err) } } + +// For a configuration that needs no special transport, rest.HTTPClientFor returns http.DefaultClient +// itself. Installing the redirect policy on it would change every HTTP caller in the process, and +// race with them. +func TestHTTPClientForLeavesTheDefaultClientAlone(t *testing.T) { + cfg := &rest.Config{Host: "http://example.invalid"} + if shared, err := rest.HTTPClientFor(cfg); err != nil || shared != http.DefaultClient { + t.Skip("client-go no longer returns http.DefaultClient here; nothing to protect") + } + before := http.DefaultClient.CheckRedirect + client, err := HTTPClientFor(cfg) + if err != nil { + t.Fatal(err) + } + if client == http.DefaultClient { + t.Fatal("HTTPClientFor returned http.DefaultClient itself") + } + if http.DefaultClient.CheckRedirect != nil || before != nil { + t.Fatal("HTTPClientFor installed its redirect policy on http.DefaultClient") + } + if client.CheckRedirect == nil { + t.Fatal("the returned client follows redirects") + } +} diff --git a/gateway/shared.go b/gateway/shared.go index cf1c88b..622f8a9 100644 --- a/gateway/shared.go +++ b/gateway/shared.go @@ -53,12 +53,16 @@ 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. +// The backoff after a shared upstream watch fails: opening it fails with UPSTREAM_UNAVAILABLE, the open +// watch dies of it, or the watch keeps ending before it is of use. 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. +// +// A watch is of use once it has completed its snapshot and stayed live for minUsefulCycle, the same +// rule the stream loop applies to one connection. Only such a watch's end resets the backoff; one +// that opens and fails at once has not recovered. One early end is tolerated, as continuity loss is +// (§5); a second in a row backs off. const ( sharedRetryMin = time.Second sharedRetryMax = 30 * time.Second @@ -93,6 +97,8 @@ type SharedBackend struct { type sharedBackoff struct { until time.Time delay time.Duration + // early counts consecutive watches that ended before they were of use. + early int } // NewSharedBackend shares one upstream watch per scope across every consumer of it. @@ -180,11 +186,14 @@ func (b *SharedBackend) recordFailureLocked(key string, err error) { } } bo := b.backoff[key] - if bo == nil { + switch { + case bo == nil: bo = &sharedBackoff{delay: sharedRetryMin} b.backoff[key] = bo - } else if bo.delay = 2 * bo.delay; bo.delay > sharedRetryMax { - bo.delay = sharedRetryMax + case bo.delay == 0: // so far only an early end, which waits for nothing + bo.delay = sharedRetryMin + default: + bo.delay = min(2*bo.delay, sharedRetryMax) } wait := bo.delay if se.RetryAfterMs != nil { @@ -223,28 +232,54 @@ 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. 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) - } +// scopeEnd is how a shared upstream watch ended, for the backoff. +type scopeEnd struct { + // useful: it completed its snapshot and stayed live for minUsefulCycle. Its end resets the backoff. + useful bool + // early: it lost continuity (a close, a 410) before it was of use. + early bool + // failure is an UPSTREAM_UNAVAILABLE it died of, if any. + failure error + cause error } -// 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) { +// forget drops a dead-or-empty scope so the next Watch opens a fresh one, and accounts for how it +// ended in the same critical section, so no Watch can slip in between and reopen the upstream +// without seeing the backoff. Only the scope's current watch is accounted for: an obsolete one ending +// late must not touch a newer attempt's backoff. +func (b *SharedBackend) forget(key string, s *sharedScope, end scopeEnd) { b.mu.Lock() defer b.mu.Unlock() - if b.scopes[key] == s { + if b.scopes[key] != s { // a newer incarnation + return + } + delete(b.scopes, key) + if end.useful { delete(b.backoff, key) } + switch { + case end.failure != nil: + b.recordFailureLocked(key, end.failure) + case end.early: + b.recordEarlyEndLocked(key, end.cause) + } +} + +// recordEarlyEndLocked counts a watch that ended before it was of use. The first is tolerated; a +// second in a row is a failing upstream, and backs off. +func (b *SharedBackend) recordEarlyEndLocked(key string, cause error) { + bo := b.backoff[key] + if bo == nil { + // until is now, not zero, so the sweep in recordFailureLocked keeps the count for a period. + b.backoff[key] = &sharedBackoff{until: b.now(), early: 1} + return + } + if bo.early++; bo.early < 2 { + return + } + unavailable := UpstreamUnavailable("the shared upstream watch keeps ending before it is of use; retry later", 0) + unavailable.Cause = cause + b.recordFailureLocked(key, unavailable) } // sharedScope is one upstream watch, its warm cache, and everyone reading from it. @@ -255,11 +290,18 @@ type sharedScope struct { watcher Watcher cancel context.CancelFunc - mu sync.Mutex - cache map[string]KRMObject // uid -> the object's current state - synced bool // has the upstream snapshot completed at least once? - dead bool - subs map[*subscriber]struct{} + mu sync.Mutex + cache map[string]KRMObject // uid -> the object's current state + synced bool // has the upstream snapshot completed at least once? + syncedAt time.Time // when it first did + dead bool + subs map[*subscriber]struct{} +} + +// usefulLocked reports whether this watch completed its snapshot and stayed live for minUsefulCycle. +// Called with s.mu held. +func (s *sharedScope) usefulLocked() bool { + return s.synced && s.backend.now().Sub(s.syncedAt) >= minUsefulCycle } // subscriber is one consumer's view of the shared watch: a SNAPSHOT, then a bounded queue of live @@ -413,15 +455,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. 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 + // everyone waiting for one can have it. + if !s.synced { + s.syncedAt = s.backend.now() + } s.synced = true for sub := range s.subs { if sub.awaiting { @@ -465,9 +507,6 @@ func (s *sharedScope) pump(ctx context.Context) { return } s.mu.Unlock() - if recovered { - s.backend.recovered(s.key, s) - } } } @@ -487,7 +526,22 @@ 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, cause) + s.mu.Lock() + useful := s.usefulLocked() + s.mu.Unlock() + end := scopeEnd{useful: useful, cause: cause} + var se *StreamError + if errors.As(cause, &se) && se != nil { + switch { + case !se.Terminal && se.Code == CodeUpstreamUnavailable: + end.failure = cause + case !se.Terminal && se.Code == CodeResyncRequired: + end.early = !useful + } + } else if cause == nil || errors.Is(cause, ErrWatchClosed) { + end.early = !useful + } + s.backend.forget(s.key, s, end) s.cancel() // Whatever ended the upstream — a clean close, a 410, a cancelled context — means one thing to a @@ -520,10 +574,12 @@ func (s *sharedScope) leave(sub *subscriber) { if empty { s.dead = true } + useful := s.usefulLocked() s.mu.Unlock() if empty { - s.backend.forget(s.key, s, nil) + // Nobody is watching any more: not a failure, but a useful watch still resets the backoff. + s.backend.forget(s.key, s, scopeEnd{useful: useful}) s.cancel() // stops pump, which stops the upstream watch } } diff --git a/gateway/shared_test.go b/gateway/shared_test.go index 7d6afec..3ae73be 100644 --- a/gateway/shared_test.go +++ b/gateway/shared_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "testing" "time" ) @@ -559,7 +560,7 @@ func TestSharedBackendBacksOffAWatchThatDiesAfterOpening(t *testing.T) { // 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) { +func TestSharedBackoffResetsOnlyAfterAUsefulWatch(t *testing.T) { healthy := false up := backendFunc(func() (Watcher, error) { if healthy { @@ -596,22 +597,29 @@ func TestSharedBackoffResetsOnlyAfterASnapshot(t *testing.T) { 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") + backingOff := func() bool { + b.mu.Lock() + defer b.mu.Unlock() + _, ok := b.backoff[key] + return ok + } + // A snapshot alone is not recovery: a watch that fails straight after it has not recovered. + if !backingOff() { + t.Fatal("the backoff was reset by a snapshot the watch did not outlive") + } + now = now.Add(minUsefulCycle) + w.Stop() // the last subscriber leaves a watch that was of use + if backingOff() { + t.Error("the backoff survived a watch that was of use") } } -// 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) { +// A watch can end after it was replaced: its pump notices last. That late end must not touch the +// replacement's backoff, or the next subscriber would reopen the upstream at once. +func TestAnObsoleteWatchsLateEndCannotEraseANewerBackoff(t *testing.T) { opens := 0 failing := false up := backendFunc(func() (Watcher, error) { @@ -619,7 +627,7 @@ func TestAnObsoleteWatchsRecoveryCannotEraseANewerBackoff(t *testing.T) { if failing { return nil, UpstreamUnavailable("the API server is unavailable", 0) } - return &stubWatcher{}, nil // opens, then idles: its recovery is replayed by hand below + return &stubWatcher{}, nil // opens, then idles: its end is replayed by hand below }) b := NewSharedBackend(up) now := time.Unix(0, 0) @@ -627,7 +635,7 @@ func TestAnObsoleteWatchsRecoveryCannotEraseANewerBackoff(t *testing.T) { scope := Scope{Version: "v1", Resource: "configmaps", Namespace: "app"} key := scopeKey(scope) - // 1. The old watch is open, and is about to report its recovery. + // 1. The old watch is open. w, err := b.Watch(t.Context(), scope) if err != nil { t.Fatal(err) @@ -645,13 +653,89 @@ func TestAnObsoleteWatchsRecoveryCannotEraseANewerBackoff(t *testing.T) { t.Fatal("the replacement opened") } - // 4. The old watch's delayed recovery arrives. - b.recovered(key, old) + // 4. The old watch's end arrives late, looking like an early end of its own. + old.die(ErrWatchClosed) 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) + t.Errorf("upstream opens = %d, want 2: the obsolete watch's end changed the newer backoff", opens) + } +} + +// The stream loop stops each connection after two early cycles, but a cohort of reconnecting +// subscribers would still reopen a shared upstream once per subscriber. The shared scope applies the +// same rule to its upstream watch: one early end is tolerated, the second backs off, and the cohort +// meets that backoff instead of the API server. +func TestASharedUpstreamThatKeepsEndingEarlyIsBackedOffForTheWholeCohort(t *testing.T) { + for name, watcher := range map[string]func() Watcher{ + "closed before its snapshot": func() Watcher { return &endsWith{err: ErrWatchClosed} }, + "410 right after its snapshot": func() Watcher { + return &endsWith{events: []WatchEvent{boundary}, err: ResyncRequired("the upstream resourceVersion expired (410 Gone)")} + }, + } { + t.Run(name, func(t *testing.T) { + var opens atomic.Int32 + b := NewSharedBackend(backendFunc(func() (Watcher, error) { opens.Add(1); return watcher(), nil })) + frozen := time.Unix(0, 0) + b.now = func() time.Time { return frozen } // the backoff never expires during the test + gw := &Gateway{Auth: AllowAll{}, Clients: func(context.Context, string, Principal) (Backend, error) { return b, nil }} + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + var wg sync.WaitGroup + for range 100 { + wg.Go(func() { + err := gw.Stream(ctx, nil, Scope{Version: "v1", Resource: "configmaps", Namespace: "app"}, &lockedSink{}) + var se *StreamError + if !errors.As(err, &se) || se.Code != CodeUpstreamUnavailable || se.Terminal { + t.Errorf("Stream() = %v, want a non-terminal UPSTREAM_UNAVAILABLE", err) + } + }) + } + wg.Wait() + if ctx.Err() != nil { + t.Fatal("the streams did not end") + } + if n := opens.Load(); n != 2 { + t.Errorf("upstream opens = %d for 100 streams, want 2: one tolerated early end, then the backoff", n) + } + }) + } +} + +// Ordinary closures of a watch that was of use (the API server's routine timeout) are recovered in +// band, as often as they come, and never back off. +func TestASharedUpstreamsLongLivedWatchesRecoverWithoutBackoff(t *testing.T) { + var opens atomic.Int32 + b := NewSharedBackend(backendFunc(func() (Watcher, error) { + if opens.Add(1) > 5 { + return nil, Forbidden("stop") // ends the test + } + return &endsWith{events: []WatchEvent{boundary}, err: ErrWatchClosed}, nil + })) + clock := steppingClock(2 * time.Second) // every watch outlives its snapshot by a second or more + b.now = clock + gw := &Gateway{Auth: AllowAll{}, now: clock, Clients: func(context.Context, string, Principal) (Backend, error) { return b, nil }} + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + err := gw.Stream(ctx, nil, Scope{Version: "v1", Resource: "configmaps", Namespace: "app"}, &lockedSink{}) + var se *StreamError + if !errors.As(err, &se) || se.Code != CodeForbidden { + t.Fatalf("Stream() = %v after %d opens, want the scripted FORBIDDEN: long-lived watches were backed off", err, opens.Load()) } } + +type lockedSink struct { + mu sync.Mutex + events []Event +} + +func (s *lockedSink) Emit(_ context.Context, ev Event) error { + s.mu.Lock() + defer s.mu.Unlock() + s.events = append(s.events, ev) + return nil +} diff --git a/spec/v1.md b/spec/v1.md index f797dff..a471859 100644 --- a/spec/v1.md +++ b/spec/v1.md @@ -270,7 +270,8 @@ stable client behavior. v1 codes: `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 unavailable upstream, or one whose watches keep ending before they are of use; 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 From d8d8052d56febaf0be59cd18841b354bf4109c51 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 2 Oct 2026 20:09:34 +0000 Subject: [PATCH 4/5] fix(gateway): pass typed errors to shared subscribers even when caused by a closed watch SharedBackend's die() classified its cause StreamError-first for the backoff, but then still converted anything matching errors.Is(cause, ErrWatchClosed) into RESYNC_REQUIRED for subscribers. StreamError.Unwrap exposes its Cause, so a terminal FORBIDDEN, or an UPSTREAM_UNAVAILABLE with a retry hint, caused by a closed watch reached the browser as a resync it retried. Decide once whether the cause is typed, and convert only an untyped close. Co-Authored-By: Claude Opus 5.5 --- gateway/shared.go | 11 +++++-- gateway/shared_test.go | 67 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 3 deletions(-) diff --git a/gateway/shared.go b/gateway/shared.go index 622f8a9..27b9bf7 100644 --- a/gateway/shared.go +++ b/gateway/shared.go @@ -530,15 +530,19 @@ func (s *sharedScope) die(cause error) { useful := s.usefulLocked() s.mu.Unlock() end := scopeEnd{useful: useful, cause: cause} + // A StreamError decides before ErrWatchClosed does, here and below: StreamError.Unwrap exposes + // its Cause, so a terminal FORBIDDEN caused by a closed watch also matches ErrWatchClosed. var se *StreamError - if errors.As(cause, &se) && se != nil { + typed := errors.As(cause, &se) && se != nil + closed := !typed && (cause == nil || errors.Is(cause, ErrWatchClosed)) + if typed { switch { case !se.Terminal && se.Code == CodeUpstreamUnavailable: end.failure = cause case !se.Terminal && se.Code == CodeResyncRequired: end.early = !useful } - } else if cause == nil || errors.Is(cause, ErrWatchClosed) { + } else if closed { end.early = !useful } s.backend.forget(s.key, s, end) @@ -547,7 +551,8 @@ func (s *sharedScope) die(cause error) { // Whatever ended the upstream — a clean close, a 410, a cancelled context — means one thing to a // subscriber: continuity is lost, start a new cycle. Say it as a non-terminal RESYNC_REQUIRED so // the stream loop announces it and resnapshots, rather than tearing the browser's connection down. - if cause == nil || errors.Is(cause, ErrWatchClosed) { + // A typed error is passed on as it is, with its code, terminal flag, message and hint. + if closed { cause = ResyncRequired("the shared upstream watch ended; a new snapshot cycle follows") } diff --git a/gateway/shared_test.go b/gateway/shared_test.go index 3ae73be..0ccc227 100644 --- a/gateway/shared_test.go +++ b/gateway/shared_test.go @@ -739,3 +739,70 @@ func (s *lockedSink) Emit(_ context.Context, ev Event) error { s.events = append(s.events, ev) return nil } + +// A typed error that ends the shared upstream reaches every subscriber as it is, even when its Cause +// is ErrWatchClosed. StreamError.Unwrap exposes that cause, and reading it as a routine close would +// turn a terminal refusal into a resync the browser retries, and drop an unavailable error's hint. +func TestSharedSubscribersReceiveTypedErrorsCausedByAClosedWatch(t *testing.T) { + forbidden := Forbidden("the gateway's identity may no longer watch this scope") + forbidden.Cause = ErrWatchClosed + unavailable := UpstreamUnavailable("the API server is restarting", 4*time.Second) + unavailable.Cause = ErrWatchClosed + for name, sent := range map[string]*StreamError{"terminal FORBIDDEN": forbidden, "UPSTREAM_UNAVAILABLE with a hint": unavailable} { + for _, asEvent := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/event=%v", name, asEvent), func(t *testing.T) { + b := NewSharedBackend(backendFunc(func() (Watcher, error) { + if asEvent { // a WatchError event on the open watch + return &endsWith{events: []WatchEvent{boundary, {Type: WatchError, Err: sent}}, err: ErrWatchClosed}, nil + } + return &endsWith{events: []WatchEvent{boundary}, err: sent}, nil // a failing Next + })) + w, err := b.Watch(t.Context(), Scope{Version: "v1", Resource: "configmaps", Namespace: "app"}) + if err != nil { + t.Fatal(err) + } + defer w.Stop() + for err == nil { + var ev WatchEvent + ev, err = w.Next(t.Context()) + if err == nil && ev.Type == WatchError { + err = ev.Err + } + } + var got *StreamError + if !errors.As(err, &got) { + t.Fatalf("subscriber got %v, want the typed error", err) + } + if got.Code != sent.Code || got.Terminal != sent.Terminal || got.Message != sent.Message || !equalHint(got.RetryAfterMs, sent.RetryAfterMs) { + t.Errorf("subscriber got %s terminal=%v %q hint=%v, want %s terminal=%v %q hint=%v", + got.Code, got.Terminal, got.Message, got.RetryAfterMs, sent.Code, sent.Terminal, sent.Message, sent.RetryAfterMs) + } + }) + } + } +} + +func equalHint(a, b *int) bool { return (a == nil) == (b == nil) && (a == nil || *a == *b) } + +// Seen from the browser: the terminal refusal is the stream's last event, not a resync and a retry. +func TestAStreamOverASharedBackendEndsWithTheTypedRefusal(t *testing.T) { + forbidden := Forbidden("the gateway's identity may no longer watch this scope") + forbidden.Cause = ErrWatchClosed + b := NewSharedBackend(backendFunc(func() (Watcher, error) { + return &endsWith{events: []WatchEvent{boundary}, err: forbidden}, nil + })) + sink, err := streamOver(t, b, nil) + var se *StreamError + if !errors.As(err, &se) || se.Code != CodeForbidden || !se.Terminal { + t.Fatalf("Stream() = %v, want the terminal FORBIDDEN", err) + } + var codes []string + for _, ev := range sink.events { + if ev.Type == EventError { + codes = append(codes, fmt.Sprintf("%s terminal=%v", ev.Code, ev.Terminal)) + } + } + if len(codes) != 1 || codes[0] != "FORBIDDEN terminal=true" { + t.Errorf("error events = %v, want only the terminal FORBIDDEN", codes) + } +} From 7978a315032592ae289013a1e6e75678aadf28f5 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 2 Oct 2026 20:12:32 +0000 Subject: [PATCH 5/5] ci: require a released adapter's core requirement to equal the release version The release check only verified that the core tag the adapter requires exists. Had a release PR left the marked requirement at an older, tagged version, the adapter-tag-only build would pass while resolving an older core. Compare the two when RELEASE_VERSION is set. (CodeRabbit review on #42.) Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bbc85e6..51e4c14 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -244,6 +244,13 @@ jobs: echo "::error file=gateway/kube/go.mod::the core requirement is a pseudo-version (v${core}). Release Please owns this line; restore the released version. CI already builds an unreleased adapter with the core from the same commit." exit 1 fi + # A release's adapter must require the core it is released with. Release Please writes that + # line; if it ever did not, the adapter would resolve an older core, and nothing else here + # would notice, because the older tag exists. + if [ -n "$RELEASE_VERSION" ] && [ "$core" != "$RELEASE_VERSION" ]; then + echo "::error file=gateway/kube/go.mod::the released adapter requires core v${core}, not the v${RELEASE_VERSION} it is released with" + exit 1 + fi # On a release PR the line names the version being released, whose tag does not exist until # the PR merges. Skip that case in ordinary CI; release CI requires the tags to exist. if ! git ls-remote --exit-code --tags origin "refs/tags/gateway/v${core}" >/dev/null 2>&1; then