From f8c3e30341b160c7c75c55da4f0ea1d6c98d7848 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 11 Sep 2026 13:37:40 +0000 Subject: [PATCH 1/4] feat!: bound HTTP delivery and balance stream lifecycle observations Implements Proposal 0007 after a scope review that cut it from four increments to three: transport correctness, a minimal set of balanced lifecycle events, and a tested host example. Identity resolution stays in the example rather than becoming public API, and the real-cluster scenario reuses `task test-cluster` instead of adding a CI workflow. Transport: add opt-in `Options.WriteTimeout` / `Gateway.WriteTimeout` bounding each HTTP write plus flush under one absolute deadline, covering headers, frames, heartbeats and terminal events. Successful operations clear the deadline so quiet streams outlive many timeout periods; a failed operation poisons the sink and cancels its stream rather than being retried. Flush errors and short writes now propagate. Add `CheckHTTPStreaming` so hosts can test a mounted middleware stack before deploying it. With a positive bound, a writer that cannot provide deadlines aborts before streaming and reports `http_transport_rejected`. No unbounded diagnostic frame is attempted, so clients may reconnect; this is a deliberate tradeoff, since an unsupported writer cannot guarantee both bounded completion and delivery of an explanation. Observations: add `stream_closed`, `shared_subscription_opened` and `shared_subscription_closed` with no new `Observation` fields. These count logical streams and active attachments. They are not HTTP request counts and not physical API-server watch counts; measure those independently. HTTP entry and authorization instrumentation stay host-owned for now. BREAKING CHANGE: `WriteSSEHeaders` is removed; use `Gateway.ServeStream` or `ServeStreamProjection`, which own headers, delivery and cleanup. `SSESink.Heartbeat` now returns an error and its caller must stop the stream on failure. `NewSSESink(io.Writer)` stays generic and installs no HTTP deadline. The v1 wire protocol is unchanged. Co-Authored-By: Claude Opus 5 (1M context) --- docs/adopting.md | 9 + docs/auth.md | 17 +- docs/operations.md | 46 ++- ...006-stream-and-save-implementation-plan.md | 5 + .../0007-shared-stream-host-integration.md | 223 +++++++++++ examples/README.md | 4 + gateway/CHANGELOG.md | 15 + gateway/README.md | 10 + gateway/handler.go | 21 +- gateway/kube/CHANGELOG.md | 7 + gateway/kube/authz.go | 4 +- gateway/kube/examples/sharedstream/README.md | 102 +++++ .../kube/examples/sharedstream/e2e_test.go | 305 +++++++++++++++ gateway/kube/examples/sharedstream/handler.go | 109 ++++++ .../examples/sharedstream/handler_test.go | 201 ++++++++++ gateway/kube/examples/sharedstream/metrics.go | 32 ++ gateway/kube/examples/sharedstream/subject.go | 35 ++ gateway/lifecycle_test.go | 106 +++++ gateway/observe.go | 25 +- gateway/shared.go | 4 +- gateway/sse.go | 235 ++++++++---- gateway/sse_test.go | 363 ++++++++++++++++++ gateway/stream.go | 4 + gateway/stream_test.go | 2 +- 24 files changed, 1797 insertions(+), 87 deletions(-) create mode 100644 docs/proposals/0007-shared-stream-host-integration.md create mode 100644 gateway/kube/examples/sharedstream/README.md create mode 100644 gateway/kube/examples/sharedstream/e2e_test.go create mode 100644 gateway/kube/examples/sharedstream/handler.go create mode 100644 gateway/kube/examples/sharedstream/handler_test.go create mode 100644 gateway/kube/examples/sharedstream/metrics.go create mode 100644 gateway/kube/examples/sharedstream/subject.go create mode 100644 gateway/lifecycle_test.go create mode 100644 gateway/sse_test.go diff --git a/docs/adopting.md b/docs/adopting.md index c1f3cfe..de8967f 100644 --- a/docs/adopting.md +++ b/docs/adopting.md @@ -149,3 +149,12 @@ boundary; a per-user backend keeps Kubernetes RBAC as the direct boundary by con Next: [saving.md](saving.md) for the host-owned write path and [operations.md](operations.md) for stream monitoring and limits. + +## Tested shared host and middleware + +Use the [compiled shared ConfigMap host](../gateway/kube/examples/sharedstream/README.md) to compose +participant identity resolution, service-account sharing, per-subscriber authorization, session +expiry and `WriteTimeout`. Its [middleware capability recipe](../gateway/kube/examples/sharedstream/README.md#middleware-capability-test) +checks the actual wrapped server writer in a host test. `CheckHTTPStreaming` must run before +streaming: it emits no response, but clears the existing write deadline. A positive timeout requires +flush and deadline support; unsupported transports abort and can cause browser reconnects. diff --git a/docs/auth.md b/docs/auth.md index 849f73a..e00fd18 100644 --- a/docs/auth.md +++ b/docs/auth.md @@ -102,7 +102,7 @@ opts.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend ``` The [`SubjectAccessReviewAuthorizer`](../gateway/kube/authz.go) adapter delegates the decision to -Kubernetes. `subjectOf` maps the principal to the Kubernetes username and groups. The adapter checks +Kubernetes. `subjectOf` supplies the API-server-resolved username, groups, UID and extras. The adapter checks both `list` and `watch`. An incomplete review is refused, and an explicit `Denied` wins over `Allowed`. @@ -116,3 +116,18 @@ The host owns writes, CSRF protection, audit and write authorization. Before a m `gateway.ValidateMergePatch` with the effective projection and current object, and include the captured UID and resourceVersion preconditions. Project any resource returned to the browser. See [saving](saving.md) for the complete flow. + +## Tested shared-host composition + +The [shared ConfigMap host](../gateway/kube/examples/sharedstream/README.md) demonstrates a local +SelfSubjectReview helper using participant credentials, service-account SARs and data access, +fixed scope, session/token expiry and bounded HTTP delivery. Identity resolution is an example, +not a public library authentication API. It does not re-resolve identity on every timed check. + +`ReauthorizationTimeout` applies to periodic callbacks after acquiring the delivery gate. Opening +and cycle authorization need their own host callback deadlines. Do not put a short callback deadline +on the entire healthy stream. Write bounds limit in-flight HTTP I/O, not all gate waiting, backend +operations or callback work; declare and measure the total revocation budget under the intended load. +For 200 allowed participants, opening can issue 400 SARs plus 200 SSRs. Timers can align, recovery +adds checks, and client-side throttling consumes callback budgets. Neither the example's rate settings +nor Voter's reported rehearsal results are production defaults or supported-version evidence. diff --git a/docs/operations.md b/docs/operations.md index 78dccd4..f26bcbf 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -16,9 +16,10 @@ object name, UID, principal, patch contents, or an error message. | Signal | Meaning | First response | |---|---|---| +| `http_transport_rejected` | bounded HTTP serving lacks required writer capabilities; no logical stream opened | test the mounted middleware and provide flush/deadline support; aborted requests may reconnect | | `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` | authorization, upstream, or protocol failure ended a stream | alert by low-cardinality error code; browsers must not retry terminal errors | +| `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 | | `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 | @@ -27,6 +28,7 @@ object name, UID, principal, patch contents, or an error message. | Control | Default | Use | |---|---:|---| +| `gateway.Options.WriteTimeout` | 0 (no deadline) | set a positive per-operation write-plus-flush budget for bounded HTTP delivery | | `gateway.Options.HeartbeatInterval` | 20 seconds | set below the shortest proxy idle timeout | | `gateway.SharedOptions.QueueDepth` | 256 live events | tune after measuring; it bounds memory per slow subscriber | | `ScopePolicy.AllowLabelSelector` | false | enable only for an endpoint that deliberately supports caller narrowing | @@ -36,3 +38,45 @@ object name, UID, principal, patch contents, or an error message. 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. Measure snapshot size and duration per allowed scope before opening broad all-namespaces endpoints. + +## Counting lifetimes + +`stream_opened`/`stream_closed` count `StreamProjection` entry/return, including authorization +failure, but exclude HTTP identity/scope refusals before entry. `shared_subscription_opened` and +`shared_subscription_closed` count active attachments, including warm-cache joins. Overflow, scope +death or leave ends an attachment once; repeated `Stop` does not count again. Cleanup may finish +later. A resnapshot can replace an attachment on one logical stream without another API watch. + +Callbacks are synchronous and may run concurrently, including under shared locks. Return promptly; +do not panic or reenter the gateway. Update gauge counters synchronously rather than through a lossy +exporter queue. Opens precede matching closes, with no global order across independent lifetimes. +Ignore unknown kinds. Scope values are not safe metric labels. See the tested +[counter mapping](../gateway/kube/examples/sharedstream/metrics.go). + +These gauges do not count HTTP requests still resolving identity or upstream watch handles. Keep +host metrics for those needs. Measure physical API-server WATCH requests independently with the +metrics available on the tested Kubernetes version; the shared-host fixture uses +`apiserver_longrunning_requests` filtered to the resource and `verb="WATCH"`. + +## Bounded HTTP delivery + +Set `WriteTimeout` explicitly; zero preserves no library-installed deadline and negative values +are rejected. Each header, event or heartbeat write and flush shares one deadline. Successful +operations clear it, so quiet streams can outlive many timeout periods. A failed operation cancels +its stream and cannot be retried as a terminal frame. Flush success does not acknowledge browser +receipt. Generic `Stream` sinks remain responsible for bounded I/O. + +The transport owns write deadlines while serving and attempts to clear them at exit. It cannot +retrieve and restore a host's previous deadline. A whole-response `http.Server.WriteTimeout` is +not a per-operation substitute for a long-lived stream. Earlier request deadlines constrain I/O; +request cancellation may still wait for the current operation's bound. Backend opening/cleanup, +authorization gate waiting, encoding and host callbacks are outside that I/O budget. + +With a positive bound, unsupported writers abort before streaming and emit `http_transport_rejected`. +No unbounded diagnostic frame is attempted; browser reconnects are possible. Test the actual mounted +middleware using the [capability-check recipe](../gateway/kube/examples/sharedstream/README.md#middleware-capability-test). +HTTP/1.1 sockets and TLS HTTP/2 (including two streams on one connection) are tested, along with +transparent `Unwrap` and opaque wrappers. Other middleware/proxy combinations need host tests. + +For authorization bursts and the limits of the 200-subscriber profile, see the +[shared-host capacity guide](../gateway/kube/examples/sharedstream/README.md#verification-and-capacity). diff --git a/docs/proposals/0006-stream-and-save-implementation-plan.md b/docs/proposals/0006-stream-and-save-implementation-plan.md index e089f73..5a988fc 100644 --- a/docs/proposals/0006-stream-and-save-implementation-plan.md +++ b/docs/proposals/0006-stream-and-save-implementation-plan.md @@ -6,6 +6,11 @@ Follow the standing [design rules](../../CONTRIBUTING.md#design-rules) and [release policy](../releasing.md). [Proposal 0005](0005-kubernetes-stream-and-save-semantics.md) explains the unresolved tradeoffs; this document owns work order and acceptance criteria. +[Proposal 0007](0007-shared-stream-host-integration.md) proposes bounded HTTP delivery, lifecycle +observations and tested shared-host composition. That work can support the authorization bounds +and continuation measurements below, but adds no dependency or acceptance gate to this plan. +Hosts may demonstrate the existing requirements with their own bounded sinks and instrumentation. + ## Baseline and order Use the [adoption guide](../adopting.md), [saving guide](../saving.md) and diff --git a/docs/proposals/0007-shared-stream-host-integration.md b/docs/proposals/0007-shared-stream-host-integration.md new file mode 100644 index 0000000..e6844f4 --- /dev/null +++ b/docs/proposals/0007-shared-stream-host-integration.md @@ -0,0 +1,223 @@ +# Proposal 0007: Shared-stream transport correctness + +**Status: implemented in the working tree; unreleased.** The transport, observation and example +increments below have landed. The real-cluster scenario in increment 3 is manual and has not been +run here, so no `docs/facts/` record exists yet. + +## Recommendation + +Centralize bounded HTTP delivery and balanced stream/subscription observations in the library. +Keep identity resolution as a tested example and use existing manual cluster tooling for capacity +measurements. The justification is consistent transport and lifecycle behavior across hosts; +line-count reduction is a possible adoption result, not an acceptance criterion. + +This responds to Voter's 2026-09-11 request against reported release 0.3.0 and the subsequent +scope review. The initial code review used `e450334`. Voter's unpublished implementation and +200-identity rehearsal have not been reproduced here. + +Follow the [design rules](../../CONTRIBUTING.md#design-rules) and +[release policy](../releasing.md). Credentials, sessions, cluster selection, allowlists, RBAC, +writes and metrics export remain host-owned. No wire or browser-store change is planned. + +| Increment | Deliver | Keep out | +|---|---|---| +| 1 | HTTP operation deadlines, capability check, transport-rejection observation and cancellation tests | New transport framework or promise of a universal revocation bound. | +| 2 | Stream close plus shared-subscription open/close observations | New observation fields, HTTP request accounting, authorization telemetry and upstream gauges. | +| 3 | Tested shared-host example with local subject resolution and capacity notes | Public identity helper, new cluster CI workflow or adopter-specific login suite. | + +[Proposal 0006](0006-stream-and-save-implementation-plan.md) remains independently deliverable. +These changes can help satisfy its existing evidence requirements, but are not additional gates. +Hosts can demonstrate bounded sinks and measurements through their own implementations. + +## 1. Bound HTTP delivery + +The current [`sse.go`](../../gateway/sse.go) discards flush errors, ignores the emit context, +and stops only the heartbeat goroutine when its write fails. Teardown waits for that goroutine. +In [`reauthorize.go`](../../gateway/reauthorize.go), downstream delivery holds the authorization +gate, and the periodic callback timeout starts after acquiring it. These paths need testing together. + +Moving `context.WithTimeout` above `gate.mu.Lock()` alone is not a teardown fix: +[`Mutex.Lock`](https://pkg.go.dev/sync#Mutex.Lock) does not accept cancellation. The budget can +expire while the goroutine still waits, and the in-flight sink write still needs to finish. +Do not ship that move as an independent bounded-revocation guarantee. For this increment, retain +`ReauthorizationTimeout` as the periodic callback budget and bound HTTP I/O at its owner. A +cancelable gate can be considered separately if measurements establish a remaining need; it +would still not terminate an arbitrary blocked sink. + +Add `WriteTimeout time.Duration` to `Options` and `Gateway` for HTTP serving only. Zero installs +no library deadline, positive values enable the bound, and negative values are configuration +errors. Validate numeric options at handler construction and before I/O for direct serving calls. +Use five seconds explicitly in the example, not as a universal default. + +Implement within the existing SSE transport and reuse it for normal streams and the identity/scope +`refuse` path in [`handler.go`](../../gateway/handler.go): + +- One serialized operation gets one absolute deadline covering write plus flush. Include initial + headers, heartbeat comments and terminal frames. Use an earlier request deadline where present. +- Clear deadlines after successful operations so quiet streams survive many timeout periods. + Attempt cleanup on exit after transport workers finish. Do not reset a failed transport to + attempt another frame. Document ownership: a previous host deadline cannot be retrieved and restored. +- Use [`http.ResponseController`](https://pkg.go.dev/net/http#ResponseController) for deadline + support and error-aware flushing. Propagate reported flush errors and short writes. An errorless + flusher cannot report every failure; successful flushing is not browser acknowledgement. +- Cancel the internal stream context on header, frame or heartbeat failure and join the heartbeat + worker. Queued operations must not resume after failure. Never abandon a goroutine in a write. +- Preserve ordinary terminal SSE refusals on supported transports, including error propagation + from their initial headers and flush. + +### Helper and compatibility decisions + +Remove exported `WriteSSEHeaders` and replace its serving-path use with an internal error-returning +header operation under the same deadline handling as frames. Update `Handler`, direct serving +entry points and refusal paths together. No compatibility shim; direct HTTP users should use +`Gateway.ServeStream` or `ServeStreamProjection`, which own headers and transport cleanup. + +Keep `NewSSESink(io.Writer)` for generic framing, including the existing buffer-based golden tests. +Its argument need not become `http.ResponseWriter`: the HTTP serving path already has that writer +and can supply controller-backed I/O through internal configuration of the same sink. There must +be one frame encoder. Generic callers receive no HTTP timeout guarantee; propagate flush errors +when their writer exposes an error-returning flush method. Change `SSESink.Heartbeat` to return +an error so its HTTP owner can cancel the stream on failure, and update callers to handle it. + +The increment's release notes must name removal of `WriteSSEHeaders`, the serving-method migration, +the `Heartbeat` return change, the new timeout/check API and error propagation. These are public +API changes under the pre-1.0 policy, even though no new wire behavior is introduced. + +### Unsupported writers + +Check deadline/flush capability before committing SSE headers or opening a subscription. The +request's writer does not exist at `Handler` construction, so middleware capability must be tested +at request time and in the host's integration tests. Transparent wrappers should implement `Unwrap`. + +Add `CheckHTTPStreaming(w http.ResponseWriter) error`, shared by bounded HTTP serving and host +tests. It checks exposed flush support through the controller-compatible wrapper chain and probes +write-deadline support by setting a zero deadline. It must not write headers/body or invoke flush. +Return an error identifying the missing capability, preserving `errors.Is(err, http.ErrNotSupported)` +for unsupported operations. Call only before streaming with no concurrent writer use: the probe +clears any existing write deadline and cannot restore it. It checks exposed capabilities, not +whether arbitrary middleware correctly implements them or whether a proxy buffers responses. + +Add a tested adoption recipe using `httptest.NewServer` with the host's actual middleware chain +around a test-only leaf handler that calls this check and records its result for assertion. Use a +real server writer, not `httptest.ResponseRecorder`, which does not supply network deadlines. +Cover transparent-wrapper success and opaque-wrapper failure. Never mount this probe as a public +production route. Real blocked-client tests remain necessary to establish deadline behavior. + +If the host requests a bound and the writer cannot provide it, abort with `http.ErrAbortHandler`. +Do not attempt an unbounded terminal frame: even a tiny frame can block during flush, so that fallback +would violate the requested guarantee. No new wire error code is needed. + +This failure can cause browser reconnects and must not be described as a clean terminal SSE refusal. +Emit `http_transport_rejected` once before aborting, using the existing observer and internal-error +code, with no raw middleware error or claim of frame delivery. This event has no corresponding +`stream_opened` and does not change a stream gauge. A broken transport cannot guarantee both bounded +completion and an explanatory frame reaching the browser. This tradeoff is deliberate, not a +solution to automatic reconnects. + +### Acceptance + +Use real HTTP/1.1 sockets, TLS HTTP/2 and controlled failing writers. Cover a non-reading peer with +enough data to exhaust buffers, quiet/active healthy streams across multiple timeout periods, +header/refusal and heartbeat-only failures, flush errors, short writes, transparent and opaque +wrappers. Verify HTTP/2 isolation on the same connection. Keep these tests in normal Go CI. + +Measure expiry, cancellation and periodic revocation during blocked delivery while another +subscriber continues. Assert handler/heartbeat completion, subscriber release and eventual last +upstream cleanup with explicit synchronization and declared timing tolerances. + +The write timeout excludes gate waiting, encoding CPU and arbitrary callbacks or backend +`Watch`/`Stop`. Shared opening currently uses a detached context under the backend mutex; do not +claim subscriber expiry bounds it. Generic sinks and host callbacks must cooperate with cancellation. +The total revocation budget still includes timer cadence, in-flight delivery, authorization and cleanup. + +## 2. Balance a small set of lifecycle observations + +Keep the existing `Observer` interface and `Observation` fields. The complete additions are below; +`http_transport_rejected` ships with increment 1, and the three lifetime events with increment 2. + +| Observation | Counting contract | +|---|---| +| `http_transport_rejected` | One pre-stream capability rejection when bounded HTTP serving cannot be supported. No logical stream opens, no close is owed, and no SSE error delivery is attempted. | +| `stream_closed` | Pair existing `stream_opened` at `StreamProjection` entry with one close on return, including authorization failure. It excludes HTTP identity/scope refusals before entry. | +| `shared_subscription_opened` | Report successful attachment to a shared scope, including a warm-cache join. This is a new event; no shared-attach observation exists today. | +| `shared_subscription_closed` | Report once when that attachment becomes inactive through overflow, scope death or leave. A later repeated `Stop` must not count again. | + +Retain cycle, overflow, resync, suppression and terminal-error signals. A resnapshot can close and +reopen an attachment on one logical stream without creating a new upstream watch. These counts +are not physical API-server watch counts, HTTP request counts or proof that cleanup has finished. + +Preserve `terminal_error` for logical-stream failure. It already occurs before `sink.Emit` in +`emitTerminal`, so it has never guaranteed frame delivery. In increment 1, clarify this in the +`observe.go` comment and the operations alert row, and add the separate transport-rejection row. +Test that unsupported writers emit exactly one rejection, no `stream_opened` or `terminal_error`, +and open no backend; separately test that a failed terminal-frame write still reports the existing +logical-stream failure. Do not count either diagnostic as a lifetime open or close. + +Callbacks remain synchronous, prompt and concurrency-safe; no lossy library queue, new IDs or +statistics framework. Guarantee open-before-close for each lifetime, not global ordering. Hosts +update counters synchronously and must not block, panic or reenter the gateway from an observer. +Review callback placement under shared locks. Ignore unknown kinds and avoid turning `Scope` +values into metric labels; correct the overly broad safe-to-export wording in `observe.go`. + +**Acceptance:** race-test concurrent/warm-cache joins, denial, revocation, immediate upstream EOF, +recovery, overflow, transport failure and repeated cleanup. Assert intermediate counts and final +balance. Add a small tested counter mapping in the operations example. + +Keep HTTP entry/exit and authorization/SAR instrumentation in host wrappers for now. Defer upstream +handle observations too: they can describe a useful library lifetime, but do not measure physical +API-server watches. This smaller change will not let Voter delete all its watch/metrics wrappers; +retain them until equivalent observations are actually provided. Revisit additions when concrete +remaining integration code demonstrates their value, rather than requiring an arbitrary adopter count. + +## 3. Test the composition and record capacity + +Add `gateway/kube/examples/sharedstream`, following the existing conditional-save example's +placement. Use a local, tested function to issue SelfSubjectReview with a supplied participant +client and copy username, groups, UID and multi-valued extras into `kube.Subject`. Reject missing +username/nil results; preserve API and cancellation errors and independent returned data. No public +resolver API, credential discovery, identity cache, impersonation or guessed OIDC mapping. + +Show one process-wide shared backend, trusted session resolution, one server-selected cluster, +fixed namespace/name policy, participant SSR, service-account SAR/data access, explicit write and +periodic-check timeouts, and the earlier session/token expiry. Keep direct writes participant-owned +and link the existing save example. Browser identity headers must not select the client or override +the resolved subject. A service-account SSR resolves the service account; it is not participant identity. + +Opening SSR and initial/cycle authorization need explicit short callback deadlines in the example: +`ReauthorizationTimeout` covers periodic checks only. Do not impose a short deadline on the whole +healthy stream. Keep backend request timeouts and session deadlines distinct. + +Update [adoption](../adopting.md), [authorization](../auth.md), [operations](../operations.md), +the [gateway README](../../gateway/README.md) and [examples index](../../examples/README.md) +when implementation lands. Explain `2 * N / interval` SARs/second for allowed checks: about 13.3 +at 200 subscribers/30 seconds, plus opening/cycle bursts. Two hundred simultaneous openings can +need 400 SARs and 200 separate SSRs. Early denial can issue fewer; periodic checks may align. +Client throttling, shared rate limiters and API-server capacity all matter. Voter's reported +100-QPS/400-burst setting is evidence for its workload, not a default. + +Compile and test the example in normal CI, including identity copying, cancellation and host +boundary tests. Reuse manual `task test-cluster`/`task cluster-facts` machinery for a focused load +scenario; add an opt-in Task entry only if needed. No new cluster CI workflow. Record actual runs +in `docs/facts/` with commit, versions, workload, proxy/protocols, limits, commands and results. +Keep general guidance in operations/auth docs and unrun scenarios clearly separate from facts. + +Exercise concurrent openings/reconnects, one revoked identity, expiry, a blocked peer, continuing +peers and final cleanup. Use independent API-server evidence for physical watches. The 200-user, +30-second recheck, five-second check and 60-second closure profile from Proposal 0006 is a declared +measurement target, not a guarantee or release-wide load-test requirement. Measure snapshot bytes +and sustained resource/latency behavior; a short rehearsal cannot establish production capacity. +Do not import Voter's login/browser suite or extend Kubernetes support from its k3s 1.31.5 result. + +## Delivery + +Review transport, observations and example changes separately; observations can proceed independently. +For runtime PRs use the [existing checks](../../CONTRIBUTING.md#test-levels): `task fixtures-check`, +`task test`, `task lint`, focused race tests in changed Go modules, and wire/browser tests for HTTP +changes. Attach a manual real-cluster run for claims about actual SSR/SAR composition; do not label +fake-client results or skipped scenarios as cluster evidence. Planning-only changes need link and +whitespace checks. + +Use the lockstep release process and document option defaults, unsupported-writer behavior, flush +errors and event meanings. No release is required just for this proposal. Consumers pin released +artifacts and remove only wrappers whose behavior they have verified is covered; credentials, +scope/session policy and their boundary tests remain in the host. diff --git a/examples/README.md b/examples/README.md index bdbf4f8..0a0a503 100644 --- a/examples/README.md +++ b/examples/README.md @@ -14,3 +14,7 @@ For host integration patterns, use these small recipes: projected reconciliation, and a host-owned Kubernetes endpoint that preserves real 409 conflicts. [Vue adapter](vue/README.md) is a copyable, typechecked composable with reactivity and cleanup tests. + +[Shared ConfigMap host](../gateway/kube/examples/sharedstream/README.md) is a compiled Go example +with participant SelfSubjectReview, service-account SARs and shared data, bounded SSE writes, +session expiry, lifecycle counters and a manually run real-cluster fixture. diff --git a/gateway/CHANGELOG.md b/gateway/CHANGELOG.md index feef84d..77fe9a8 100644 --- a/gateway/CHANGELOG.md +++ b/gateway/CHANGELOG.md @@ -1,5 +1,20 @@ # Changelog +## Unreleased + +- 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/README.md b/gateway/README.md index 3fe7f8f..7afc1b4 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -103,3 +103,13 @@ For shared streams, set `Options.ReauthorizationInterval` (for example, 30 secon including during quiet periods; denial or timeout stops only that subscriber. Zero interval keeps cycle-only checks. See [authorization](../docs/auth.md) for callback contracts and capacity planning. Use `kube.SubjectAccessReviewAuthorizer` for Kubernetes-backed subscriber authorization. + +`Options.WriteTimeout` bounds each HTTP header/frame/comment write plus flush. Zero installs no +library deadline; the [shared-host example](kube/examples/sharedstream/README.md) chooses five seconds. +An unsupported writer aborts with `http_transport_rejected` before a logical stream opens. Use +`CheckHTTPStreaming` in a mounted middleware test; it clears the write deadline without writing. + +`stream_closed` balances logical stream opens; `shared_subscription_opened` and +`shared_subscription_closed` describe active attachments, not upstream watch counts. The observer +is synchronous and concurrent: update counters promptly and do not reenter the gateway. +See [operations](../docs/operations.md) for exact counting and timeout contracts. diff --git a/gateway/handler.go b/gateway/handler.go index 8fb3b4b..af9434b 100644 --- a/gateway/handler.go +++ b/gateway/handler.go @@ -69,6 +69,10 @@ type Options struct { // timeout; it affects HTTP streams only. HeartbeatInterval time.Duration + // WriteTimeout bounds each HTTP write-plus-flush operation. Zero installs no deadline. + // Positive values require a writer supporting flush and write deadlines; negative values panic. + WriteTimeout time.Duration + // ReauthorizationInterval rechecks each subscriber independently, even on quiet streams. // Zero disables timed checks; snapshot cycles always reauthorize. ReauthorizationInterval time.Duration @@ -84,9 +88,12 @@ type Options struct { // the only thing that works: a browser's EventSource cannot read the body of a non-200, so a 403 // reaches the page as an `onerror` with no detail and no reason, and the developer is left guessing. // A terminal error event carries a code and a message the UI can actually show, and `terminal` is -// what stops EventSource from reconnecting forever. +// what stops EventSource from reconnecting forever. Unsupported bounded transports instead +// abort before streaming and report ObservationHTTPTransportRejected. func Handler(o Options) http.Handler { switch { + case o.WriteTimeout < 0: + panic("krm-stream: WriteTimeout must not be negative") case o.Principal == nil: panic("krm-stream: Options.Principal is required — the library must never assume who the caller is") case o.Authorizer == nil: @@ -103,6 +110,7 @@ func Handler(o Options) http.Handler { Ordering: o.Ordering, Observer: o.Observer, HeartbeatInterval: o.HeartbeatInterval, + WriteTimeout: o.WriteTimeout, ReauthorizationInterval: o.ReauthorizationInterval, ReauthorizationTimeout: o.ReauthorizationTimeout, } @@ -113,7 +121,7 @@ func Handler(o Options) http.Handler { // 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. - refuse(w, Forbidden("not authenticated")) + g.refuse(w, r, Forbidden("not authenticated")) return } @@ -122,7 +130,7 @@ func Handler(o Options) http.Handler { err = o.Scopes.Validate(scope) } if err != nil { - refuse(w, asStreamError(err)) + g.refuse(w, r, asStreamError(err)) return } @@ -131,7 +139,8 @@ func Handler(o Options) http.Handler { } // refuse writes a terminal error as a well-formed one-event stream, and closes. -func refuse(w http.ResponseWriter, serr *StreamError) { - WriteSSEHeaders(w) - _ = (&sequenceSink{sink: NewSSESink(w)}).Emit(context.Background(), serr.Event()) +func (g *Gateway) refuse(w http.ResponseWriter, r *http.Request, serr *StreamError) { + g.serveHTTP(w, r, func(ctx context.Context, sink *SSESink) { + _ = (&sequenceSink{sink: sink}).Emit(ctx, serr.Event()) + }) } diff --git a/gateway/kube/CHANGELOG.md b/gateway/kube/CHANGELOG.md index f3ac6d8..dbc9d15 100644 --- a/gateway/kube/CHANGELOG.md +++ b/gateway/kube/CHANGELOG.md @@ -1,5 +1,12 @@ # 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. 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/authz.go b/gateway/kube/authz.go index 9baba9e..0a61660 100644 --- a/gateway/kube/authz.go +++ b/gateway/kube/authz.go @@ -38,9 +38,9 @@ import ( // it is whatever your session says a user is — so translating it into a Kubernetes subject is a thing // only the host can do, and this is where it does it. type Subject struct { - // User is the username RBAC binds against (the OIDC `username` claim, typically). + // User is the API-server-resolved username, including any configured mapping. User string - // Groups are the groups RBAC binds against (the OIDC `groups` claim, typically). + // Groups are the API-server-resolved groups, not guessed OIDC claim values. Groups []string // UID and Extra are optional, and are carried through verbatim for audit and for authorizers // (webhook, OPA) that key on them. diff --git a/gateway/kube/examples/sharedstream/README.md b/gateway/kube/examples/sharedstream/README.md new file mode 100644 index 0000000..2574e1b --- /dev/null +++ b/gateway/kube/examples/sharedstream/README.md @@ -0,0 +1,102 @@ +# Shared ConfigMap stream host + +This compiled example centralizes HTTP delivery and sharing while the host supplies authentication, +credentials and policy. Copy it into your application; its helper functions are not a public kube +identity API. It has no login-service dependency. + +Construct `Handler` once at process startup with a service-account `rest.Config`, a fixed namespace +and ConfigMap name, a trusted session resolver, and optionally `&Counters{}`. Mount the returned +handler at the application's stream route. The browser requests: + +```text +/resource-stream/v1?version=v1&resource=configmaps&namespace=app&name=coffee +``` + +The resolver reads the host's authenticated session, validates it, and supplies a participant token, +session expiry and token expiry. It must honor the request context. Browser-supplied identity or +API-server headers are never used by the example. Keep the route behind the host's existing session +protection; do not copy the integration test's fixture-session query routing into production. + +The handler uses the server/TLS settings from the configured cluster for all three operations: + +- Participant SelfSubjectReview resolves username, groups, UID and extras at opening. Using the + service client here would resolve the service account instead. A resolved subject is not a token + refresh or an authorization grant. The example copies the response data independently. +- Service-account SubjectAccessReviews check list and watch on the fixed scope for that subject, + before cache disclosure and every 30 seconds. Each check has a five-second callback budget, + including opening/cycle checks. Changed account/session policy should be checked by the host too. +- One shared service-account backend watches the ConfigMap. A five-second HTTP operation timeout + bounds each downstream write plus flush. The earlier session/token expiry ends the request. + +The service account needs named ConfigMap list/watch and create on `subjectaccessreviews`. +Participants need named list/watch to subscribe; direct GET/PATCH remains participant-authenticated +and separately authorized. Compose with [the conditional-save example](../conditionalsave/handler.go) +and its host CSRF/audit requirements. No write is routed through the shared service identity. + +`Counters` updates logical-stream and active-subscription gauges synchronously and counts transport +capability rejections. Unknown observation kinds are ignored. No names, subjects or scope values +become labels. HTTP requests still resolving identity and physical API watch handles are different +lifetimes; keep host instrumentation for them when needed. + +## Middleware capability test + +Use this pattern in a host test with the **actual middleware chain** mounted around the probe: + +```go +checked := make(chan error, 1) +probe := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + checked <- gateway.CheckHTTPStreaming(w) + w.WriteHeader(http.StatusNoContent) +}) +server := httptest.NewServer(yourMiddleware(probe)) +defer server.Close() +response, err := server.Client().Get(server.URL) +if err != nil { t.Fatal(err) } +_ = response.Body.Close() +if err := <-checked; err != nil { t.Fatal(err) } +``` + +Do not use `httptest.ResponseRecorder` to establish network-deadline support. Do not expose a +production probe endpoint. `CheckHTTPStreaming` writes and flushes nothing, but clears the current +write deadline; call before streaming with no concurrent writer access. It verifies exposed +capabilities, not middleware correctness or proxy buffering. Transparent wrappers need `Unwrap`. +The gateway's [mounted-wrapper test](../../../sse_test.go) exercises success and failure on real +server writers. The gateway's socket tests separately verify stalled-peer behavior. + +## Verification and capacity + +Normal `go test ./...` in `gateway/kube` compiles this example, checks subject ownership/API failures +and cancellation, and proves that forged browser identity headers cannot change its identity or +scope. The tests use a real HTTP fake API for the credential boundary; they are not Kubernetes RBAC +proof. Run the real-cluster test separately against a disposable cluster: + +```bash +# From gateway/kube, with the disposable cluster's KUBECONFIG selected: +go test -tags e2e -run '^TestSharedHostRealAPI$' -v -count=1 -timeout 4m ./examples/sharedstream +KRM_SHARED_SUBSCRIBERS=200 go test -tags e2e -run '^TestSharedHostRealAPI$' -v -count=1 -timeout 4m ./examples/sharedstream +``` + +`task test-cluster` also includes the two-identity case. No new cluster CI workflow is required. +The fixture creates isolated RBAC and independent service-account participant tokens, verifies +participant writes, opening/reconnects, one revoked identity with continuing peers, balanced host +gauges, and API-server WATCH requests returning to baseline. It requires administrative fixture +permissions and `/metrics` access; it cleans up its namespace and cluster RBAC. It measures +`apiserver_longrunning_requests` for ConfigMap WATCH requests independently of the library. +Other activity on that resource can invalidate the comparison; use an isolated fixture cluster. + +The fixture uses client QPS 100/burst 400 explicitly. At 200 allowed subscriptions and a 30-second +interval, steady authorization is about `2 * 200 / 30 = 13.3` SARs/second, plus opening/recovery +bursts. Simultaneous opening can require 400 SARs and 200 SSRs. Periodic timers may align too. +These rates are a workload description, not universal defaults. Per-client and shared rate limiters, +API-server limits and check latency must be considered together. A larger burst can simply move +contention to the API server. + +Record real runs in `docs/facts/` with commit, versions, protocols/proxy, workload, limits, commands, +results and unrun cases. The 30-second recheck, five-second check and 60-second closure target is +not a library guarantee. Gate waiting, in-flight I/O, callback execution and cleanup contribute to +revocation time. A write timeout does not bound backend opening or uncooperative callbacks. + +A shared watch still transfers a snapshot to each client and consumes per-client memory, connection +and browser work. Measure sustained load, peak memory/CPU, snapshot bytes and latency distributions +before making production capacity claims. This fixture does not run 200 browser renderers or logins; +expiry and blocked-peer isolation are covered separately by the core HTTP tests. diff --git a/gateway/kube/examples/sharedstream/e2e_test.go b/gateway/kube/examples/sharedstream/e2e_test.go new file mode 100644 index 0000000..f8b33a2 --- /dev/null +++ b/gateway/kube/examples/sharedstream/e2e_test.go @@ -0,0 +1,305 @@ +//go:build e2e + +package sharedstream + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/ConfigButler/krm-stream/gateway" + authenticationv1 "k8s.io/api/authentication/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + "k8s.io/utils/ptr" +) + +// TestSharedHostRealAPI is manual evidence, included by task test-cluster. Set +// KRM_SHARED_SUBSCRIBERS=200 for the opening/reconnect burst profile. No login +// system or browser credentials are synthesized; the fixture supplies trusted sessions. +func TestSharedHostRealAPI(t *testing.T) { + count := 2 + if value := os.Getenv("KRM_SHARED_SUBSCRIBERS"); value != "" { + n, err := strconv.Atoi(value) + if err != nil || n < 2 || n > 200 { + t.Fatal("KRM_SHARED_SUBSCRIBERS must be 2..200") + } + count = n + } + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Minute) + defer cancel() + cfg, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(clientcmd.NewDefaultClientConfigLoadingRules(), nil).ClientConfig() + if err != nil { + t.Fatal(err) + } + cfg.QPS = 100 + cfg.Burst = 400 // Declared fixture capacity, not a library default. + admin, err := kubernetes.NewForConfig(cfg) + if err != nil { + t.Fatal(err) + } + ns, err := admin.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{GenerateName: "krm-shared-"}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = admin.CoreV1().Namespaces().Delete(context.Background(), ns.Name, metav1.DeleteOptions{}) }) + _, err = admin.CoreV1().ConfigMaps(ns.Name).Create(ctx, &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "coffee"}, Data: map[string]string{"value": "initial"}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + makeToken := func(name string) string { + t.Helper() + _, err := admin.CoreV1().ServiceAccounts(ns.Name).Create(ctx, &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: name}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + token, err := admin.CoreV1().ServiceAccounts(ns.Name).CreateToken(ctx, name, &authenticationv1.TokenRequest{Spec: authenticationv1.TokenRequestSpec{ExpirationSeconds: ptr.To(int64(3600))}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + return token.Status.Token + } + serviceToken := makeToken("stream-host") + for _, name := range []string{"shared-reader", "participant"} { + verbs := []string{"list", "watch"} + if name == "participant" { + verbs = append(verbs, "get", "patch") + } + _, err := admin.RbacV1().Roles(ns.Name).Create(ctx, &rbacv1.Role{ObjectMeta: metav1.ObjectMeta{Name: name}, Rules: []rbacv1.PolicyRule{{APIGroups: []string{""}, Resources: []string{"configmaps"}, ResourceNames: []string{"coffee"}, Verbs: verbs}}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + } + bind := func(name, role string) { + t.Helper() + _, err := admin.RbacV1().RoleBindings(ns.Name).Create(ctx, &rbacv1.RoleBinding{ObjectMeta: metav1.ObjectMeta{Name: name}, RoleRef: rbacv1.RoleRef{APIGroup: "rbac.authorization.k8s.io", Kind: "Role", Name: role}, Subjects: []rbacv1.Subject{{Kind: "ServiceAccount", Name: name, Namespace: ns.Name}}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + } + bind("stream-host", "shared-reader") + _, err = admin.RbacV1().ClusterRoles().Create(ctx, &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{Name: ns.Name}, Rules: []rbacv1.PolicyRule{{APIGroups: []string{"authorization.k8s.io"}, Resources: []string{"subjectaccessreviews"}, Verbs: []string{"create"}}}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = admin.RbacV1().ClusterRoles().Delete(context.Background(), ns.Name, metav1.DeleteOptions{}) + }) + _, err = admin.RbacV1().ClusterRoleBindings().Create(ctx, &rbacv1.ClusterRoleBinding{ObjectMeta: metav1.ObjectMeta{Name: ns.Name}, RoleRef: rbacv1.RoleRef{APIGroup: "rbac.authorization.k8s.io", Kind: "ClusterRole", Name: ns.Name}, Subjects: []rbacv1.Subject{{Kind: "ServiceAccount", Namespace: ns.Name, Name: "stream-host"}}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = admin.RbacV1().ClusterRoleBindings().Delete(context.Background(), ns.Name, metav1.DeleteOptions{}) + }) + tokens := make([]string, count) + for i := range tokens { + name := fmt.Sprintf("viewer-%d", i) + tokens[i] = makeToken(name) + bind(name, "participant") + } + serviceCfg := rest.AnonymousClientConfig(cfg) + serviceCfg.BearerToken = serviceToken + counters := &Counters{} + h, err := Handler(serviceCfg, ns.Name, "coffee", func(r *http.Request) (Session, error) { + // Fixture-only routing to trusted sessions. Production resolves an authenticated cookie. + i, err := strconv.Atoi(r.URL.Query().Get("fixture_session")) + if err != nil || i < 0 || i >= count { + return Session{}, fmt.Errorf("unknown fixture session") + } + return Session{Token: tokens[i], SessionExpiry: time.Now().Add(2 * time.Minute), TokenExpiry: time.Now().Add(time.Hour)}, nil + }, counters) + if err != nil { + t.Fatal(err) + } + baseline := activeConfigMapWatches(ctx, t, admin) + host := httptest.NewServer(h) + defer host.Close() + type stream struct { + body *http.Response + events chan gateway.Event + done chan struct{} + } + streams := make([]stream, count) + open := func(i int) error { + request, err := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%s/?version=v1&resource=configmaps&namespace=%s&name=coffee&fixture_session=%d", host.URL, ns.Name, i), nil) + if err != nil { + return err + } + res, err := host.Client().Do(request) + if err != nil { + return err + } + st := stream{body: res, events: make(chan gateway.Event, 32), done: make(chan struct{})} + streams[i] = st + go func() { + defer close(st.done) + defer close(st.events) + scanner := bufio.NewScanner(res.Body) + for scanner.Scan() { + if line := scanner.Text(); strings.HasPrefix(line, "data: ") { + var ev gateway.Event + if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err != nil { + return + } + select { + case st.events <- ev: + case <-ctx.Done(): + return + } + } + } + }() + return nil + } + defer func() { + for _, s := range streams { + if s.body != nil { + _ = s.body.Body.Close() + } + } + }() + wait := func(i int, kind gateway.EventType) { + t.Helper() + for { + select { + case ev, ok := <-streams[i].events: + if !ok { + t.Fatalf("stream %d ended before %s", i, kind) + } + if ev.Type == kind { + return + } + if ev.Type == gateway.EventError { + t.Fatalf("stream %d: %+v", i, ev) + } + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + } + } + started := time.Now() + var wg sync.WaitGroup + for i := range count { + wg.Go(func() { + if err := open(i); err != nil { + t.Error(err) + } + }) + } + wg.Wait() + if t.Failed() { + return + } + for i := range count { + wait(i, gateway.EventSynced) + } + t.Logf("%d independent identities synced in %s", count, time.Since(started)) + active := activeConfigMapWatches(ctx, t, admin) + if active != baseline+1 { + t.Fatalf("API watch count baseline=%g active=%g", baseline, active) + } + t.Logf("API-server active ConfigMap WATCH requests: baseline=%g active=%g", baseline, active) + // Reconnect a quarter of the cohort while others retain the shared watch. + for i := range max(1, count/4) { + _ = streams[i].body.Body.Close() + <-streams[i].done + if err := open(i); err != nil { + t.Fatal(err) + } + wait(i, gateway.EventSynced) + } + if got := activeConfigMapWatches(ctx, t, admin); got != active { + t.Fatalf("reconnect changed watch count: %g", got) + } + participantCfg := rest.AnonymousClientConfig(cfg) + participantCfg.BearerToken = tokens[1] + participant, err := kubernetes.NewForConfig(participantCfg) + if err != nil { + t.Fatal(err) + } + patch := func(value string) { + t.Helper() + _, err := participant.CoreV1().ConfigMaps(ns.Name).Patch(ctx, "coffee", types.MergePatchType, []byte(fmt.Sprintf(`{"data":{"value":%q}}`, value)), metav1.PatchOptions{}) + if err != nil { + t.Fatal(err) + } + } + patch("updated") + for i := range count { + wait(i, gateway.EventModified) + } + started = time.Now() + if err := admin.RbacV1().RoleBindings(ns.Name).Delete(ctx, "viewer-0", metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + wait(0, gateway.EventError) + select { + case <-streams[0].done: + case <-time.After(60 * time.Second): + t.Fatal("revoked stream did not close") + } + if time.Since(started) > 60*time.Second { + t.Fatal("revocation exceeded reference budget") + } + t.Logf("revoked identity closed in %s; %d peers retained", time.Since(started), count-1) + patch("after-revocation") + for i := 1; i < count; i++ { + wait(i, gateway.EventModified) + } + for i := range streams { + _ = streams[i].body.Body.Close() + <-streams[i].done + } + deadline := time.Now().Add(10 * time.Second) + for { + got := activeConfigMapWatches(ctx, t, admin) + if got == baseline && counters.Streams.Load() == 0 && counters.Subscriptions.Load() == 0 { + break + } + if time.Now().After(deadline) { + t.Fatalf("cleanup: API=%g baseline=%g streams=%d subscriptions=%d", got, baseline, counters.Streams.Load(), counters.Subscriptions.Load()) + } + time.Sleep(100 * time.Millisecond) + } + t.Log("final API watch count and host gauges returned to baseline") +} + +func activeConfigMapWatches(ctx context.Context, t *testing.T, admin kubernetes.Interface) float64 { + t.Helper() + raw, err := admin.CoreV1().RESTClient().Get().AbsPath("/metrics").DoRaw(ctx) + if err != nil { + t.Fatal(err) + } + total := float64(0) + found := false + for line := range strings.SplitSeq(string(raw), "\n") { + if strings.HasPrefix(line, "apiserver_longrunning_requests{") && strings.Contains(line, `verb="WATCH"`) && strings.Contains(line, `resource="configmaps"`) && strings.Contains(line, `group=""`) { + found = true + fields := strings.Fields(line) + value, err := strconv.ParseFloat(fields[len(fields)-1], 64) + if err != nil { + t.Fatal(err) + } + total += value + } + } + if !found { + t.Fatal("API-server ConfigMap WATCH metric missing; cannot establish physical watch counts") + } + return total +} diff --git a/gateway/kube/examples/sharedstream/handler.go b/gateway/kube/examples/sharedstream/handler.go new file mode 100644 index 0000000..ceba5ed --- /dev/null +++ b/gateway/kube/examples/sharedstream/handler.go @@ -0,0 +1,109 @@ +// Package sharedstream is a copyable host example, not a library authentication API. +package sharedstream + +import ( + "context" + "fmt" + "net/http" + "time" + + "github.com/ConfigButler/krm-stream/gateway" + "github.com/ConfigButler/krm-stream/gateway/kube" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" +) + +// Session comes from the host's trusted session store, never from browser identity +// headers. The host must reject invalid/revoked sessions and supply both expiries. +type Session struct { + Token string + SessionExpiry time.Time + TokenExpiry time.Time +} + +type sessionKey struct{} +type resolvedSession struct { + session Session + err error +} + +// Handler constructs one shared backend for one fixed ConfigMap on one cluster. +// Call once at process startup. cluster supplies service credentials and server/TLS +// settings. Participant clients use only those server/TLS settings plus the trusted +// session token. The session callback must honor r.Context and own authentication. +// Direct GET/PATCH routes must independently use participant credentials. +func Handler(cluster *rest.Config, namespace, name string, sessionFor func(*http.Request) (Session, error), observer gateway.Observer) (http.Handler, error) { + if cluster == nil || namespace == "" || name == "" || sessionFor == nil { + return nil, fmt.Errorf("sharedstream: cluster, fixed scope and session resolver required") + } + service, err := kubernetes.NewForConfig(cluster) + if err != nil { + return nil, err + } + data, err := dynamic.NewForConfig(cluster) + if err != nil { + return nil, err + } + shared := gateway.NewSharedBackendWithOptions(kube.NewBackend(data), gateway.SharedOptions{Observer: observer}) + sar := kube.SubjectAccessReviewAuthorizer(service, func(p gateway.Principal) (kube.Subject, error) { + subject, ok := p.(kube.Subject) + if !ok { + return kube.Subject{}, fmt.Errorf("missing resolved subject") + } + return subject, nil + }) + participantConfig := rest.AnonymousClientConfig(cluster) + // AnonymousClientConfig strips credential-bearing transports and impersonation. + // This client is used for one bounded SSR, not the long-lived shared watch. + participantConfig.Timeout = 5 * time.Second + handler := gateway.Handler(gateway.Options{ + Principal: func(r *http.Request) (gateway.Principal, error) { + state := r.Context().Value(sessionKey{}).(resolvedSession) + if state.err != nil { + return nil, state.err + } + cfg := rest.CopyConfig(participantConfig) + cfg.BearerToken = state.session.Token + client, err := kubernetes.NewForConfig(cfg) + if err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + return resolveSubject(ctx, client) + }, + Authorizer: gateway.AuthorizerFunc(func(ctx context.Context, p gateway.Principal, s gateway.Scope) error { + if s.Target != "" || s.Group != "" || s.Version != "v1" || s.Resource != "configmaps" || s.Namespace != namespace || s.Name != name || s.LabelSelector != "" { + return gateway.Forbidden("outside the fixed stream scope") + } + // Applies to initial/cycle checks too; the periodic setting alone does not. + check, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + return sar.Authorize(check, p, s) + }), + Clients: func(context.Context, string, gateway.Principal) (gateway.Backend, error) { return shared, nil }, + Scopes: gateway.ScopePolicy{Targets: []string{""}, Resources: []gateway.GroupResource{{Resource: "configmaps", Scope: gateway.ResourceScopeNamespaced}}}, + Projection: gateway.ProjectionFull, + WriteTimeout: 5 * time.Second, ReauthorizationInterval: 30 * time.Second, ReauthorizationTimeout: 5 * time.Second, + Observer: observer, + }) + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + session, err := sessionFor(r) + expiry := session.SessionExpiry + if session.TokenExpiry.Before(expiry) { + expiry = session.TokenExpiry + } + if err == nil && (session.Token == "" || session.SessionExpiry.IsZero() || session.TokenExpiry.IsZero() || !expiry.After(time.Now())) { + err = fmt.Errorf("invalid or expired session") + } + ctx := r.Context() + if err == nil { + var cancel context.CancelFunc + ctx, cancel = context.WithDeadline(ctx, expiry) + defer cancel() + } + ctx = context.WithValue(ctx, sessionKey{}, resolvedSession{session: session, err: err}) + handler.ServeHTTP(w, r.WithContext(ctx)) + }), nil +} diff --git a/gateway/kube/examples/sharedstream/handler_test.go b/gateway/kube/examples/sharedstream/handler_test.go new file mode 100644 index 0000000..e10e5c9 --- /dev/null +++ b/gateway/kube/examples/sharedstream/handler_test.go @@ -0,0 +1,201 @@ +package sharedstream + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync" + "testing" + "time" + + "github.com/ConfigButler/krm-stream/gateway" + authenticationv1 "k8s.io/api/authentication/v1" + authorizationv1 "k8s.io/api/authorization/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/rest" + clienttesting "k8s.io/client-go/testing" +) + +func TestResolveSubjectOwnershipAndErrors(t *testing.T) { + info := authenticationv1.UserInfo{Username: "mapped:alice", UID: "uid", Groups: []string{"a", "b"}, Extra: map[string]authenticationv1.ExtraValue{"key": {"one", "two"}}} + for _, scenario := range []string{"success", "empty", "nil", "failure"} { + t.Run(scenario, func(t *testing.T) { + result := &authenticationv1.SelfSubjectReview{Status: authenticationv1.SelfSubjectReviewStatus{UserInfo: info}} + apiErr := errors.New("API unavailable") + client := fake.NewClientset() + client.PrependReactor("create", "selfsubjectreviews", func(clienttesting.Action) (bool, runtime.Object, error) { + switch scenario { + case "empty": + return true, &authenticationv1.SelfSubjectReview{}, nil + case "nil": + return true, nil, nil + case "failure": + return true, nil, apiErr + default: + return true, result, nil + } + }) + subject, err := resolveSubject(t.Context(), client) + if scenario != "success" { + if err == nil { + t.Fatal("expected error") + } + if scenario == "failure" && !errors.Is(err, apiErr) { + t.Fatal(err) + } + return + } + if subject.User != info.Username || subject.UID != info.UID || !reflect.DeepEqual(subject.Groups, info.Groups) || !reflect.DeepEqual([]string(subject.Extra["key"]), []string(info.Extra["key"])) { + t.Fatal(subject) + } + subject.Groups[0] = "changed" + subject.Extra["key"][0] = "changed" + if result.Status.UserInfo.Groups[0] != "a" || result.Status.UserInfo.Extra["key"][0] != "one" { + t.Fatal("subject aliases API response") + } + }) + } +} + +func TestResolveSubjectInflightCancellation(t *testing.T) { + entered := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + close(entered) + <-r.Context().Done() + })) + defer srv.Close() + client, err := kubernetes.NewForConfig(&rest.Config{Host: srv.URL}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { _, err := resolveSubject(ctx, client); done <- err }() + <-entered + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("SSR did not cancel") + } +} + +func TestHostBoundaryUsesResolvedSubjectAndFixedScope(t *testing.T) { + var mu sync.Mutex + var subjects []authorizationv1.SubjectAccessReviewSpec + var tokens []string + var watches int + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasSuffix(r.URL.Path, "selfsubjectreviews"): + mu.Lock() + tokens = append(tokens, r.Header.Get("Authorization")) + mu.Unlock() + _ = json.NewEncoder(w).Encode(&authenticationv1.SelfSubjectReview{Status: authenticationv1.SelfSubjectReviewStatus{UserInfo: authenticationv1.UserInfo{Username: "resolved:user", UID: "resolved:uid", Groups: []string{"resolved:group"}, Extra: map[string]authenticationv1.ExtraValue{"x": {"a", "b"}}}}}) + case strings.HasSuffix(r.URL.Path, "subjectaccessreviews"): + var sar authorizationv1.SubjectAccessReview + if err := json.NewDecoder(r.Body).Decode(&sar); err != nil { + t.Error(err) + return + } + mu.Lock() + subjects = append(subjects, sar.Spec) + tokens = append(tokens, r.Header.Get("Authorization")) + mu.Unlock() + sar.Status.Allowed = true + _ = json.NewEncoder(w).Encode(&sar) + case strings.HasSuffix(r.URL.Path, "configmaps"): + mu.Lock() + watches++ + tokens = append(tokens, r.Header.Get("Authorization")) + mu.Unlock() + if r.URL.Query().Get("fieldSelector") != "metadata.name=coffee" { + t.Error("missing named scope") + } + _, _ = io.WriteString(w, "{\"type\":\"BOOKMARK\",\"object\":{\"apiVersion\":\"v1\",\"kind\":\"ConfigMap\",\"metadata\":{\"resourceVersion\":\"1\",\"annotations\":{\"k8s.io/initial-events-end\":\"true\"}}}}\n") + w.(http.Flusher).Flush() + <-r.Context().Done() + default: + http.NotFound(w, r) + } + })) + defer api.Close() + counters := &Counters{} + h, err := Handler(&rest.Config{Host: api.URL, BearerToken: "service", ContentConfig: rest.ContentConfig{ContentType: "application/json"}}, "app", "coffee", func(*http.Request) (Session, error) { + return Session{Token: "participant", SessionExpiry: time.Now().Add(150 * time.Millisecond), TokenExpiry: time.Now().Add(time.Hour)}, nil + }, counters) + if err != nil { + t.Fatal(err) + } + host := httptest.NewServer(h) + defer host.Close() + for _, name := range []string{"outside", "coffee"} { + req, err := http.NewRequestWithContext(t.Context(), "GET", host.URL+"/?version=v1&resource=configmaps&namespace=app&name="+name, nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("X-Remote-User", "attacker") + req.Header.Set("Authorization", "Bearer attacker") + res, err := host.Client().Do(req) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(res.Body) + _ = res.Body.Close() + if err != nil { + t.Fatal(err) + } + if name == "outside" && !strings.Contains(string(body), "FORBIDDEN") { + t.Fatalf("scope escaped: %s", body) + } + if name == "coffee" && !strings.Contains(string(body), "synced") { + t.Fatalf("no snapshot: %s", body) + } + } + mu.Lock() + defer mu.Unlock() + if watches != 1 || len(subjects) != 2 { + t.Fatalf("watches=%d SARs=%d", watches, len(subjects)) + } + for _, s := range subjects { + if s.User != "resolved:user" || s.UID != "resolved:uid" || s.Groups[0] != "resolved:group" || len(s.Extra["x"]) != 2 || s.ResourceAttributes.Name != "coffee" { + t.Fatal(s) + } + } + if strings.Join(tokens, ",") != "Bearer participant,Bearer participant,Bearer service,Bearer service,Bearer service" { + t.Fatal(tokens) + } + if counters.Streams.Load() != 0 || counters.Subscriptions.Load() != 0 { + t.Fatal("unbalanced lifetimes") + } +} + +func TestCounterMappingConcurrentAndUnknown(t *testing.T) { + c := &Counters{} + var wg sync.WaitGroup + for range 100 { + wg.Go(func() { + for _, kind := range []gateway.ObservationKind{gateway.ObservationStreamOpened, gateway.ObservationSharedSubscriptionOpened, "future_kind", gateway.ObservationSharedSubscriptionClosed, gateway.ObservationStreamClosed} { + c.Observe(gateway.Observation{Kind: kind}) + } + }) + } + wg.Wait() + c.Observe(gateway.Observation{Kind: gateway.ObservationHTTPTransportRejected}) + if c.Streams.Load() != 0 || c.Subscriptions.Load() != 0 || c.TransportRejected.Load() != 1 { + t.Fatal("counter mapping is unbalanced") + } +} diff --git a/gateway/kube/examples/sharedstream/metrics.go b/gateway/kube/examples/sharedstream/metrics.go new file mode 100644 index 0000000..e27d460 --- /dev/null +++ b/gateway/kube/examples/sharedstream/metrics.go @@ -0,0 +1,32 @@ +package sharedstream + +import ( + "sync/atomic" + + "github.com/ConfigButler/krm-stream/gateway" +) + +// Counters illustrates a low-cardinality Observer mapping. Export these atomics +// through the host's metrics endpoint. None measure physical API-server watches. +// The same observer can be used by both the gateway and its shared backend. +type Counters struct { + Streams atomic.Int64 + Subscriptions atomic.Int64 + TransportRejected atomic.Int64 +} + +// Observe updates gauges synchronously; unknown kinds are intentionally ignored. +func (c *Counters) Observe(o gateway.Observation) { + switch o.Kind { + case gateway.ObservationStreamOpened: + c.Streams.Add(1) + case gateway.ObservationStreamClosed: + c.Streams.Add(-1) + case gateway.ObservationSharedSubscriptionOpened: + c.Subscriptions.Add(1) + case gateway.ObservationSharedSubscriptionClosed: + c.Subscriptions.Add(-1) + case gateway.ObservationHTTPTransportRejected: + c.TransportRejected.Add(1) + } +} diff --git a/gateway/kube/examples/sharedstream/subject.go b/gateway/kube/examples/sharedstream/subject.go new file mode 100644 index 0000000..2753408 --- /dev/null +++ b/gateway/kube/examples/sharedstream/subject.go @@ -0,0 +1,35 @@ +package sharedstream + +import ( + "context" + "fmt" + "slices" + + "github.com/ConfigButler/krm-stream/gateway/kube" + authenticationv1 "k8s.io/api/authentication/v1" + authorizationv1 "k8s.io/api/authorization/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" +) + +// The supplied client MUST authenticate as the participant. A service-account +// client resolves that service account instead. This is neither token refresh nor +// an authorization grant; the host chooses when the captured identity expires. +func resolveSubject(ctx context.Context, participant kubernetes.Interface) (kube.Subject, error) { + result, err := participant.AuthenticationV1().SelfSubjectReviews().Create(ctx, &authenticationv1.SelfSubjectReview{}, metav1.CreateOptions{}) + if err != nil { + return kube.Subject{}, fmt.Errorf("resolve Kubernetes subject: %w", err) + } + if result == nil || result.Status.UserInfo.Username == "" { + return kube.Subject{}, fmt.Errorf("SelfSubjectReview returned no username") + } + info := result.Status.UserInfo + subject := kube.Subject{User: info.Username, UID: info.UID, Groups: slices.Clone(info.Groups)} + if info.Extra != nil { + subject.Extra = make(map[string]authorizationv1.ExtraValue, len(info.Extra)) + for key, values := range info.Extra { + subject.Extra[key] = slices.Clone(authorizationv1.ExtraValue(values)) + } + } + return subject, nil +} diff --git a/gateway/lifecycle_test.go b/gateway/lifecycle_test.go new file mode 100644 index 0000000..57bfe5c --- /dev/null +++ b/gateway/lifecycle_test.go @@ -0,0 +1,106 @@ +package gateway + +import ( + "context" + "errors" + "sync" + "testing" +) + +type lifecycleCounter struct { + mu sync.Mutex + counts map[ObservationKind]int + negative bool +} + +func (c *lifecycleCounter) Observe(o Observation) { + c.mu.Lock() + defer c.mu.Unlock() + if c.counts == nil { + c.counts = make(map[ObservationKind]int) + } + c.counts[o.Kind]++ + if c.counts[ObservationSharedSubscriptionClosed] > c.counts[ObservationSharedSubscriptionOpened] || c.counts[ObservationStreamClosed] > c.counts[ObservationStreamOpened] { + c.negative = true + } +} +func (c *lifecycleCounter) assert(t *testing.T, open, closed int) { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + if c.negative || c.counts[ObservationSharedSubscriptionOpened] != open || c.counts[ObservationSharedSubscriptionClosed] != closed { + t.Fatalf("lifetimes = %v, negative=%v", c.counts, c.negative) + } +} + +func TestSharedSubscriptionLifecycle(t *testing.T) { + b := newFakeUpstream() + counter := &lifecycleCounter{} + shared := NewSharedBackendWithOptions(b, SharedOptions{QueueDepth: 1, Observer: counter}) + first, err := shared.Watch(t.Context(), sharedScopeUnderTest) + if err != nil { + t.Fatal(err) + } + b.send(WatchEvent{Type: WatchBookmark, InitialEventsEnd: true}) + drainSnapshot(t, first) + var wg sync.WaitGroup + const joins = 20 + for range joins { + wg.Go(func() { + w, err := shared.Watch(t.Context(), sharedScopeUnderTest) + if err != nil { + t.Error(err) + return + } + w.Stop() + w.Stop() + }) + } + wg.Wait() + counter.assert(t, joins+1, joins) + if b.opens() != 1 { + t.Fatal("warm joins reopened upstream") + } + // Overflow ends the attachment before the owner calls Stop. + sw := first.(*sharedWatcher) + sw.scope.mu.Lock() + sw.sub.offer(WatchEvent{Type: WatchAdded, Object: obj("a", "a", "1")}) + sw.sub.offer(WatchEvent{Type: WatchAdded, Object: obj("b", "b", "2")}) + sw.scope.mu.Unlock() + counter.assert(t, joins+1, joins+1) + first.Stop() + first.Stop() + counter.assert(t, joins+1, joins+1) +} + +func TestScopeDeathAndFailedAttachmentLifetimes(t *testing.T) { + counter := &lifecycleCounter{} + shared := NewSharedBackendWithOptions(newFakeUpstream(), SharedOptions{Observer: counter}) + w, err := shared.Watch(t.Context(), sharedScopeUnderTest) + if err != nil { + t.Fatal(err) + } + s := w.(*sharedWatcher).scope + s.die(ErrWatchClosed) + counter.assert(t, 1, 1) + if _, err := s.subscribe(); !errors.Is(err, ErrWatchClosed) { + t.Fatal(err) + } + w.Stop() + s.die(ErrWatchClosed) + counter.assert(t, 1, 1) +} + +type failedOpenBackend struct{} + +func (failedOpenBackend) Watch(context.Context, Scope) (Watcher, error) { + return nil, errors.New("open failed") +} +func TestFailedSharedOpenHasNoLifetime(t *testing.T) { + counter := &lifecycleCounter{} + b := NewSharedBackendWithOptions(failedOpenBackend{}, SharedOptions{Observer: counter}) + if _, err := b.Watch(t.Context(), sharedScopeUnderTest); err == nil { + t.Fatal("expected failure") + } + counter.assert(t, 0, 0) +} diff --git a/gateway/observe.go b/gateway/observe.go index 3760e49..20ca3fa 100644 --- a/gateway/observe.go +++ b/gateway/observe.go @@ -6,8 +6,19 @@ package gateway type ObservationKind string const ( - // ObservationStreamOpened reports a newly accepted stream request. + // ObservationStreamOpened reports StreamProjection entry, before authorization. + // HTTP identity/scope refusals before entry are excluded. ObservationStreamOpened ObservationKind = "stream_opened" + // ObservationStreamClosed pairs each logical stream open with return after cleanup. + ObservationStreamClosed ObservationKind = "stream_closed" + // ObservationHTTPTransportRejected reports a failed bounded-HTTP capability check. + // No logical stream opens and no SSE delivery is attempted. + ObservationHTTPTransportRejected ObservationKind = "http_transport_rejected" + // ObservationSharedSubscriptionOpened reports an attachment, including warm-cache joins. + ObservationSharedSubscriptionOpened ObservationKind = "shared_subscription_opened" + // ObservationSharedSubscriptionClosed reports an attachment becoming inactive once. + // Overflow, scope death and leave end attachment; owned cleanup may finish later. + ObservationSharedSubscriptionClosed ObservationKind = "shared_subscription_closed" // ObservationCycleStarted reports a new snapshot cycle after authorization and projection selection. ObservationCycleStarted ObservationKind = "cycle_started" // ObservationEventEmitted reports a protocol event accepted by the consumer sink. @@ -18,11 +29,13 @@ const ( ObservationConsumerResync ObservationKind = "consumer_resync" // ObservationSharedOverflow reports a shared-watch subscriber exceeding its bounded queue. ObservationSharedOverflow ObservationKind = "shared_overflow" - // ObservationTerminalError reports a stream-ending error code. + // ObservationTerminalError reports a logical-stream failure before attempting its + // terminal frame. It does not guarantee delivery. ObservationTerminalError ObservationKind = "terminal_error" ) -// Observation is one safe-to-export stream lifecycle signal. +// Observation describes a lifecycle signal. Scope contains high-cardinality data +// and must not be exported blindly as metric labels. type Observation struct { Kind ObservationKind Scope Scope @@ -32,7 +45,11 @@ type Observation struct { } // Observer receives stream lifecycle signals. Observe must return promptly; it runs on the stream or -// shared-watch goroutine. Use it to increment counters or enqueue into a non-blocking exporter. +// shared-watch goroutine, sometimes under shared locks. Calls are synchronous and +// may be concurrent; they are not dropped by the library. Observers must be thread-safe, +// must not panic or reenter the gateway, and should update counters synchronously. +// Each lifetime opens before closing, with no global ordering across lifetimes. +// A lossy exporter queue must not be the source of truth for lifetime gauges. type Observer interface { Observe(Observation) } diff --git a/gateway/shared.go b/gateway/shared.go index fc71216..20d0c3b 100644 --- a/gateway/shared.go +++ b/gateway/shared.go @@ -57,7 +57,7 @@ type SharedOptions struct { // QueueDepth is the maximum number of live events a subscriber may lag before it is resnapshotted. // Zero uses 256. QueueDepth int - // Observer receives low-cardinality overflow signals. It must not block. + // Observer receives subscription-lifecycle and overflow signals. It must not block. Observer Observer } @@ -239,6 +239,7 @@ func (sub *subscriber) end(reason error) { return } sub.closed = true + sub.scope.backend.observe(Observation{Kind: ObservationSharedSubscriptionClosed, Scope: sub.scope.scope}) sub.reason = reason close(sub.ch) } @@ -260,6 +261,7 @@ func (s *sharedScope) subscribe() (Watcher, error) { awaiting: true, } s.subs[sub] = struct{}{} + s.backend.observe(Observation{Kind: ObservationSharedSubscriptionOpened, Scope: s.scope}) // The warm cache, and the entire point of the exercise: if the upstream snapshot is already // complete, this consumer gets its whole reset…synced now, from memory, and the API server never diff --git a/gateway/sse.go b/gateway/sse.go index fc0a15c..38197f5 100644 --- a/gateway/sse.go +++ b/gateway/sse.go @@ -2,6 +2,7 @@ package gateway import ( "context" + "errors" "fmt" "io" "net/http" @@ -29,118 +30,210 @@ import ( // idle timeout of every proxy anyone actually deploys behind. const HeartbeatInterval = 20 * time.Second -// SSESink writes protocol events to a connection as Server-Sent Events. It is safe for the heartbeat -// goroutine and the stream loop to use concurrently — which they must, since the whole point of a -// heartbeat is that it happens while nothing else is. +// SSESink frames events for an io.Writer. Calls are serialized. Generic writers have +// no library-installed deadline; HTTP callers should use Gateway.ServeStream. type SSESink struct { - mu sync.Mutex - w io.Writer - flush func() + mu sync.Mutex + w io.Writer + flush func() error + before func(context.Context) error + after func() error + cancel context.CancelFunc + failed error } -// NewSSESink writes to w, flushing after every frame if w can be flushed. +// NewSSESink writes to w, flushing after every frame when supported. A FlushError +// method takes precedence over http.Flusher so reported failures reach the caller. func NewSSESink(w io.Writer) *SSESink { - s := &SSESink{w: w, flush: func() {}} - if f, ok := w.(http.Flusher); ok { - s.flush = f.Flush + s := &SSESink{w: w, flush: func() error { return nil }} + switch f := w.(type) { + case interface{ FlushError() error }: + s.flush = f.FlushError + case http.Flusher: + s.flush = func() error { f.Flush(); return nil } } return s } -// Emit writes one event and flushes. -func (s *SSESink) Emit(_ context.Context, ev Event) error { - frame, err := ev.MarshalSSE() - if err != nil { - return fmt.Errorf("krm-stream: marshal event: %w", err) - } +// operation serializes the entire deadline/write/flush/clear operation. Failures +// poison the sink so a queued heartbeat or terminal event cannot revive it. +func (s *SSESink) operation(ctx context.Context, write func() error) (err error) { s.mu.Lock() defer s.mu.Unlock() - if _, err := s.w.Write(frame); err != nil { + if s.failed != nil { + return s.failed + } + defer func() { + if err != nil { + s.failed = err + if s.cancel != nil { + s.cancel() + } + } + }() + if err = ctx.Err(); err != nil { + return err + } + if s.before != nil { + if err = s.before(ctx); err != nil { + return err + } + } + if err = write(); err != nil { return err } - s.flush() + if err = s.flush(); err != nil { + return err + } + if s.after != nil { + return s.after() + } return nil } -// Comment writes an SSE comment. Comments are NOT events: a conforming consumer ignores them -// entirely, which is exactly why a heartbeat is one. -func (s *SSESink) Comment(text string) error { - s.mu.Lock() - defer s.mu.Unlock() - if _, err := fmt.Fprintf(s.w, ": %s\n\n", text); err != nil { +func (s *SSESink) write(ctx context.Context, frame []byte) error { + return s.operation(ctx, func() error { + n, err := s.w.Write(frame) + if err == nil && n != len(frame) { + return io.ErrShortWrite + } return err + }) +} + +// Emit writes and flushes one event. +func (s *SSESink) Emit(ctx context.Context, ev Event) error { + frame, err := ev.MarshalSSE() + if err != nil { + return fmt.Errorf("krm-stream: marshal event: %w", err) } - s.flush() - return nil + return s.write(ctx, frame) } -// Heartbeat keeps the connection alive until ctx is done. Run it in a goroutine alongside the stream. -func (s *SSESink) Heartbeat(ctx context.Context, every time.Duration) { +// Comment writes an SSE comment, not a protocol event. +func (s *SSESink) Comment(text string) error { + return s.write(context.Background(), []byte(": "+text+"\n\n")) +} + +// Heartbeat runs until cancellation or an I/O failure. Its owner must handle the +// returned error and stop the stream on failure. Nonpositive intervals use the default. +func (s *SSESink) Heartbeat(ctx context.Context, every time.Duration) error { + if every <= 0 { + every = HeartbeatInterval + } t := time.NewTicker(every) defer t.Stop() for { select { case <-ctx.Done(): - return + return ctx.Err() case <-t.C: - if err := s.Comment("heartbeat"); err != nil { - return // the consumer went away; the stream loop will notice too + if err := s.write(ctx, []byte(": heartbeat\n\n")); err != nil { + return err } } } } -// WriteSSEHeaders sets the response headers a conforming stream must carry, and must be called -// before the first frame. -func WriteSSEHeaders(w http.ResponseWriter) { - h := w.Header() - h.Set("Content-Type", "text/event-stream") - h.Set("X-KRM-Stream-Protocol", fmt.Sprint(ProtocolVersion)) - // A stream that a proxy is allowed to cache, buffer or transform is not a stream. `no-cache` - // stops the browser; `no-transform` and the nginx-specific hint stop the middleboxes that - // otherwise sit on the response until it is "big enough" — which turns a live status watch into - // a batch job nobody can debug. - h.Set("Cache-Control", "no-cache, no-transform") - h.Set("Connection", "keep-alive") - h.Set("X-Accel-Buffering", "no") - w.WriteHeader(http.StatusOK) - if f, ok := w.(http.Flusher); ok { - f.Flush() +// CheckHTTPStreaming checks flush and write-deadline capabilities without writing +// or flushing a response. It follows Unwrap methods like http.ResponseController. +// The deadline probe clears any existing write deadline. Call before streaming, +// with no concurrent use of w. This checks exposed capabilities, not middleware +// correctness or proxy buffering. Unsupported operations wrap http.ErrNotSupported. +func CheckHTTPStreaming(w http.ResponseWriter) error { + for current := w; ; { + switch v := current.(type) { + case interface{ FlushError() error }: + return checkWriteDeadline(w) + case http.Flusher: + return checkWriteDeadline(w) + case interface{ Unwrap() http.ResponseWriter }: + current = v.Unwrap() + default: + return fmt.Errorf("krm-stream: HTTP flush: %w", http.ErrNotSupported) + } } } -// ServeStream runs one stream over one HTTP response, heartbeats included, and returns when the -// stream ends — at which point the caller returning from its handler closes the connection, which is -// what a terminal error requires. +func checkWriteDeadline(w http.ResponseWriter) error { + if err := http.NewResponseController(w).SetWriteDeadline(time.Time{}); err != nil { + return fmt.Errorf("krm-stream: HTTP write deadline: %w", err) + } + return nil +} + +func writeSSEHeaders(ctx context.Context, w http.ResponseWriter, s *SSESink) error { + return s.operation(ctx, func() error { + h := w.Header() + h.Set("Content-Type", "text/event-stream") + h.Set("X-KRM-Stream-Protocol", fmt.Sprint(ProtocolVersion)) + h.Set("Cache-Control", "no-cache, no-transform") + h.Set("Connection", "keep-alive") + h.Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + return nil + }) +} + +// ServeStream runs one HTTP stream, including headers, heartbeats and cleanup. func (g *Gateway) ServeStream(w http.ResponseWriter, r *http.Request, principal Principal, scope Scope) { g.ServeStreamProjection(w, r, principal, scope, "") } -// ServeStreamProjection writes a stream using a caller-requested projection name. Projection -// authorization happens inside StreamProjection so direct callers and HTTP share the same rule. +// ServeStreamProjection serves a requested projection under the gateway's policy. +// With WriteTimeout enabled, unsupported writers abort before a stream opens. func (g *Gateway) ServeStreamProjection(w http.ResponseWriter, r *http.Request, principal Principal, scope Scope, projection Projection) { - WriteSSEHeaders(w) + g.serveHTTP(w, r, func(ctx context.Context, sink *SSESink) { + // A terminal frame may itself fail; delivery is not guaranteed on a failed transport. + _ = g.StreamProjection(ctx, principal, scope, projection, sink) + }) +} +func (g *Gateway) serveHTTP(w http.ResponseWriter, r *http.Request, run func(context.Context, *SSESink)) { + if g.WriteTimeout < 0 { + panic("krm-stream: WriteTimeout must not be negative") + } + if g.WriteTimeout > 0 { + if err := CheckHTTPStreaming(w); err != nil { + g.observe(Observation{Kind: ObservationHTTPTransportRejected, Code: CodeInternal}) + panic(http.ErrAbortHandler) + } + } + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() sink := NewSSESink(w) - ctx := r.Context() - - hb, stopHeartbeat := context.WithCancel(ctx) - interval := g.HeartbeatInterval - if interval <= 0 { - interval = HeartbeatInterval + sink.cancel = cancel + controller := http.NewResponseController(w) + // Zero timeout retains optional flushing, including transparent HTTP wrappers. + sink.flush = func() error { + err := controller.Flush() + if g.WriteTimeout == 0 && errors.Is(err, http.ErrNotSupported) { + return nil + } + return err } - heartbeatDone := make(chan struct{}) + if g.WriteTimeout > 0 { + sink.before = func(ctx context.Context) error { + deadline := time.Now().Add(g.WriteTimeout) + if earlier, ok := ctx.Deadline(); ok && earlier.Before(deadline) { + deadline = earlier + } + return controller.SetWriteDeadline(deadline) + } + sink.after = func() error { return controller.SetWriteDeadline(time.Time{}) } + // Runs after the heartbeat has exited. Clearing never revives a failed response. + defer func() { _ = controller.SetWriteDeadline(time.Time{}) }() + } + if err := writeSSEHeaders(ctx, w, sink); err != nil { + return + } + done := make(chan struct{}) go func() { - defer close(heartbeatDone) - sink.Heartbeat(hb, interval) - }() - defer func() { - stopHeartbeat() - <-heartbeatDone + defer close(done) + if err := sink.Heartbeat(ctx, g.HeartbeatInterval); err != nil { + cancel() + } }() - - // The error is already ON the wire by the time Stream returns — emitting it is how the consumer - // learns anything. There is nothing left to tell the HTTP layer: the status line went out with - // the very first byte, long before we could have known. - _ = g.StreamProjection(ctx, principal, scope, projection, sink) + defer func() { cancel(); <-done }() + run(ctx, sink) } diff --git a/gateway/sse_test.go b/gateway/sse_test.go new file mode 100644 index 0000000..e406b72 --- /dev/null +++ b/gateway/sse_test.go @@ -0,0 +1,363 @@ +package gateway + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +type transparentWriter struct{ http.ResponseWriter } + +func (w transparentWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter } + +type opaqueWriter struct{ http.ResponseWriter } + +func TestCheckHTTPStreamingMountedMiddleware(t *testing.T) { + for _, opaque := range []bool{false, true} { + t.Run(fmt.Sprint(opaque), func(t *testing.T) { + checked := make(chan error, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if opaque { + w = opaqueWriter{w} + } else { + w = transparentWriter{w} + } + checked <- CheckHTTPStreaming(w) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + res, err := server.Client().Get(server.URL) + if err != nil { + t.Fatal(err) + } + _ = res.Body.Close() + err = <-checked + if opaque != errors.Is(err, http.ErrNotSupported) { + t.Fatalf("capability error = %v", err) + } + }) + } + if err := CheckHTTPStreaming(httptest.NewRecorder()); !errors.Is(err, http.ErrNotSupported) { + t.Fatal(err) + } +} + +func TestUnsupportedTransportHasNoStreamLifetime(t *testing.T) { + var observations []Observation + g := &Gateway{WriteTimeout: time.Second, Observer: ObserverFunc(func(o Observation) { observations = append(observations, o) })} + func() { + defer func() { + if got := recover(); got != http.ErrAbortHandler { + t.Fatalf("panic = %v", got) + } + }() + g.ServeStream(httptest.NewRecorder(), httptest.NewRequest("GET", "/", nil), nil, Scope{}) + }() + if len(observations) != 1 || observations[0].Kind != ObservationHTTPTransportRejected { + t.Fatalf("observations = %v", observations) + } +} + +type failingHTTPWriter struct { + header http.Header + failFlush int + flushes, writes int + deadlines []time.Time + short bool +} + +func (w *failingHTTPWriter) Header() http.Header { return w.header } +func (*failingHTTPWriter) WriteHeader(int) {} +func (w *failingHTTPWriter) Write(p []byte) (int, error) { + w.writes++ + if w.short { + return len(p) - 1, nil + } + return len(p), nil +} +func (w *failingHTTPWriter) FlushError() error { + w.flushes++ + if w.flushes == w.failFlush { + return io.ErrClosedPipe + } + return nil +} +func (w *failingHTTPWriter) SetWriteDeadline(d time.Time) error { + w.deadlines = append(w.deadlines, d) + return nil +} + +type transportBackend struct { + events chan WatchEvent + stopped chan struct{} + once sync.Once +} + +func newTransportBackend() *transportBackend { + return &transportBackend{events: make(chan WatchEvent, 8), stopped: make(chan struct{})} +} +func (b *transportBackend) Watch(context.Context, Scope) (Watcher, error) { return b, nil } +func (b *transportBackend) Next(ctx context.Context) (WatchEvent, error) { + select { + case ev := <-b.events: + return ev, nil + case <-ctx.Done(): + return WatchEvent{}, ctx.Err() + } +} +func (b *transportBackend) Stop() { b.once.Do(func() { close(b.stopped) }) } +func transportGateway(b Backend) *Gateway { + return &Gateway{Auth: AllowAll{}, Clients: func(context.Context, string, Principal) (Backend, error) { return b, nil }, WriteTimeout: 300 * time.Millisecond, HeartbeatInterval: 20 * time.Millisecond} +} +func waitTransport(t *testing.T, ch <-chan struct{}) { + t.Helper() + select { + case <-ch: + case <-time.After(8 * time.Second): + t.Fatal("transport cleanup did not complete") + } +} + +func TestHTTPFailuresCancelQuietStream(t *testing.T) { + for _, fail := range []int{1, 2, 3} { // headers, reset, heartbeat (no upstream events) + t.Run(fmt.Sprint(fail), func(t *testing.T) { + b := newTransportBackend() + w := &failingHTTPWriter{header: make(http.Header), failFlush: fail} + g := transportGateway(b) + done := make(chan struct{}) + go func() { + defer close(done) + g.ServeStream(w, httptest.NewRequest("GET", "/", nil), nil, sharedScopeUnderTest) + }() + waitTransport(t, done) + if fail > 1 { + waitTransport(t, b.stopped) + } + if w.flushes != fail { + t.Fatalf("wrote after failure: %d flushes", w.flushes) + } + if !w.deadlines[len(w.deadlines)-1].IsZero() { + t.Fatal("deadline not cleared on exit") + } + }) + } +} + +func TestSinkShortWriteAndFailedTerminalDelivery(t *testing.T) { + w := &failingHTTPWriter{header: make(http.Header), short: true} + s := NewSSESink(w) + if err := s.Comment("x"); !errors.Is(err, io.ErrShortWrite) { + t.Fatal(err) + } + _ = s.Comment("again") + if w.writes != 1 || w.flushes != 0 { + t.Fatal("failed sink was reused") + } + var kinds []ObservationKind + g := &Gateway{Auth: AuthorizerFunc(func(context.Context, Principal, Scope) error { return Forbidden("denied") }), Observer: ObserverFunc(func(o Observation) { kinds = append(kinds, o.Kind) })} + if err := g.Stream(t.Context(), nil, Scope{}, s); !errors.Is(err, io.ErrShortWrite) { + t.Fatal(err) + } + if fmt.Sprint(kinds) != "[stream_opened terminal_error stream_closed]" { + t.Fatal(kinds) + } +} + +// Exhaust real socket / HTTP2 stream flow-control buffers. The healthy HTTP2 +// subscriber shares the same client transport and connection with the stalled one. +func TestBlockedHTTPSubscriberReleasesSharedWatch(t *testing.T) { + for _, h2 := range []bool{false, true} { + for _, trigger := range []string{"write-timeout", "expiry", "revocation"} { + t.Run(fmt.Sprintf("http2=%t/%s", h2, trigger), func(t *testing.T) { + b := newTransportBackend() + shared := NewSharedBackend(b) + g := transportGateway(shared) + g.WriteTimeout = 2 * time.Second + g.ReauthorizationInterval = 20 * time.Millisecond + g.ReauthorizationTimeout = 2 * time.Second + var revoked atomic.Bool + g.Auth = AuthorizerFunc(func(_ context.Context, p Principal, _ Scope) error { + if p == "slow" && revoked.Load() { + return Forbidden("revoked") + } + return nil + }) + slowDone, fastDone := make(chan struct{}), make(chan struct{}) + slowCtx, expire := context.WithCancel(t.Context()) + defer expire() + var conns atomic.Int32 + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + who := strings.TrimPrefix(r.URL.Path, "/") + if who == "slow" { + defer close(slowDone) + r = r.WithContext(slowCtx) + } else { + defer close(fastDone) + } + g.ServeStream(w, r, who, sharedScopeUnderTest) + })) + srv.EnableHTTP2 = h2 + srv.Config.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateNew { + conns.Add(1) + } + } + if h2 { + srv.StartTLS() + } else { + srv.Start() + } + defer srv.Close() + client := srv.Client() + client.Timeout = 5 * time.Second + fast, err := client.Get(srv.URL + "/fast") + if err != nil { + t.Fatal(err) + } + defer func() { _ = fast.Body.Close() }() + fastEvents := make(chan string, 16) + // The test closes this body itself once isolation is proven; the read error that + // causes is the test ending, not the healthy stream failing. + var fastClosing atomic.Bool + readerDone := make(chan struct{}) + go func() { + defer close(readerDone) + scanner := bufio.NewScanner(fast.Body) + scanner.Buffer(make([]byte, 4096), 32<<20) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "data:") { + fastEvents <- line + } + } + if err := scanner.Err(); err != nil && !fastClosing.Load() { + t.Errorf("healthy reader: %v", err) + } + close(fastEvents) + }() + if h2 { + slow, err := client.Get(srv.URL + "/slow") + if err != nil { + t.Fatal(err) + } + defer func() { _ = slow.Body.Close() }() + if slow.ProtoMajor != 2 { + t.Fatal("HTTP2 not negotiated") + } + } else { + conn, err := net.Dial("tcp", srv.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = conn.Close() }() + if tcp, ok := conn.(*net.TCPConn); ok { + _ = tcp.SetReadBuffer(1024) + } + if _, err := fmt.Fprintf(conn, "GET /slow HTTP/1.1\r\nHost: test\r\n\r\n"); err != nil { + t.Fatal(err) + } + } + // Wait for both attachments before publishing the snapshot. + deadline := time.Now().Add(time.Second) + for { + shared.mu.Lock() + ss := shared.scopes[scopeKey(sharedScopeUnderTest)] + count := 0 + if ss != nil { + ss.mu.Lock() + count = len(ss.subs) + ss.mu.Unlock() + } + shared.mu.Unlock() + if count == 2 { + break + } + if time.Now().After(deadline) { + t.Fatal("subscribers did not attach") + } + time.Sleep(time.Millisecond) + } + big := obj("big", "large", "1") + big["data"] = map[string]any{"payload": strings.Repeat("x", 16<<20)} + b.events <- WatchEvent{Type: WatchAdded, Object: big} + b.events <- WatchEvent{Type: WatchBookmark, InitialEventsEnd: true} + // Start the trigger after delivery has begun; the frame exceeds socket and H2 buffers. + waitFast := func(marker string) { + t.Helper() + for { + select { + case line, ok := <-fastEvents: + if !ok { + t.Fatal("healthy peer closed") + } + if strings.Contains(line, marker) { + return + } + case <-time.After(8 * time.Second): + t.Fatalf("healthy peer missing %s", marker) + } + } + } + waitFast(`"type":"added"`) + start := time.Now() + if trigger == "expiry" { + expire() + } + if trigger == "revocation" { + revoked.Store(true) + } + waitTransport(t, slowDone) + t.Logf("%s to slow handler return: %s (write budget %s; 8s test tolerance)", trigger, time.Since(start), g.WriteTimeout) + b.events <- WatchEvent{Type: WatchAdded, Object: obj("later", "still-live", "2")} + waitFast("still-live") + if h2 && conns.Load() != 1 { + t.Fatalf("isolation used %d connections", conns.Load()) + } + fastClosing.Store(true) + _ = fast.Body.Close() + waitTransport(t, fastDone) + waitTransport(t, b.stopped) + // Join the reader so it cannot report into a finished subtest. + waitTransport(t, readerDone) + }) + } + } +} + +func TestHealthyHTTPStreamOutlivesWriteTimeout(t *testing.T) { + b := newTransportBackend() + g := transportGateway(b) + g.WriteTimeout = 40 * time.Millisecond + // Idle longer than a write period between heartbeats: stale deadlines would kill it. + g.HeartbeatInterval = 120 * time.Millisecond + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { g.ServeStream(w, r, nil, sharedScopeUnderTest) })) + defer srv.Close() + client := srv.Client() + client.Timeout = 2 * time.Second + res, err := client.Get(srv.URL) + if err != nil { + t.Fatal(err) + } + defer func() { _ = res.Body.Close() }() + reader := bufio.NewReader(res.Body) + n := 0 + for n < 3 { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatal(err) + } + if strings.HasPrefix(line, ": heartbeat") { + n++ + } + } +} diff --git a/gateway/stream.go b/gateway/stream.go index 627b6c5..11db932 100644 --- a/gateway/stream.go +++ b/gateway/stream.go @@ -90,6 +90,9 @@ type Gateway struct { // 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 + // WriteTimeout bounds each HTTP write-plus-flush operation. Zero installs no deadline. + // It does not bound generic Stream sinks, callbacks or time waiting for delivery locks. + WriteTimeout time.Duration // ReauthorizationInterval rechecks each subscriber independently, even on quiet streams. // Zero disables timed checks; snapshot cycles always reauthorize. @@ -117,6 +120,7 @@ func (g *Gateway) Stream(ctx context.Context, principal Principal, scope Scope, func (g *Gateway) StreamProjection(ctx context.Context, principal Principal, scope Scope, requested Projection, sink Sink) error { sink = &sequenceSink{sink: sink} g.observe(Observation{Kind: ObservationStreamOpened, Scope: scope}) + defer g.observe(Observation{Kind: ObservationStreamClosed, Scope: scope}) // Redaction revisions are per connection. Unlike suppression state they survive resync cycles, so // the first snapshot after a gap can report a withheld value that changed while disconnected. revisions := map[string]map[string]redactionState{} diff --git a/gateway/stream_test.go b/gateway/stream_test.go index f0be5af..32551e2 100644 --- a/gateway/stream_test.go +++ b/gateway/stream_test.go @@ -227,7 +227,7 @@ func TestGatewayObserverReportsTerminalError(t *testing.T) { if !errors.As(err, &streamErr) || streamErr.Code != CodeForbidden { t.Fatalf("Stream() error = %v, want terminal FORBIDDEN", err) } - if !sawObservation(observations, ObservationTerminalError, "") || observations[len(observations)-1].Code != CodeForbidden { + if !sawObservation(observations, ObservationTerminalError, "") || observations[len(observations)-2].Code != CodeForbidden || observations[len(observations)-1].Kind != ObservationStreamClosed { t.Fatalf("terminal observation = %#v, want FORBIDDEN", observations) } } From ed5c4d0281e287a7f73a37537e59babebade36fb Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 11 Sep 2026 13:43:19 +0000 Subject: [PATCH 2/4] test(kube): give the shared-host cluster scenario exclusive watch measurement TestSharedHostRealAPI reads apiserver_longrunning_requests, a cluster-wide gauge with no namespace label, so every other ConfigMap watcher is counted too. Under `go test ./...` the backend e2e package runs concurrently and its own real watches inflated the reading, failing the final cleanup assertion at API=26 against a baseline of 23. The library was never at fault: run alone, the scenario shows exactly one upstream watch for two subscribers and returns to baseline on cleanup. Serialize the cluster packages with -p 1 and say in the test that it needs exclusive use of the cluster. Co-Authored-By: Claude Opus 5 (1M context) --- Taskfile.yml | 4 +++- gateway/kube/examples/sharedstream/e2e_test.go | 5 ++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/Taskfile.yml b/Taskfile.yml index f921a4a..80ecc94 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -434,4 +434,6 @@ tasks: env: KUBECONFIG: '{{.KUBECONFIG_PATH}}' cmds: - - go test -tags e2e -v -count=1 -timeout 10m ./... + # -p 1: the shared-host example measures a CLUSTER-WIDE API-server watch gauge, which the + # backend suite's own real watches corrupt if their packages run concurrently. + - go test -tags e2e -v -count=1 -p 1 -timeout 10m ./... diff --git a/gateway/kube/examples/sharedstream/e2e_test.go b/gateway/kube/examples/sharedstream/e2e_test.go index f8b33a2..350aec8 100644 --- a/gateway/kube/examples/sharedstream/e2e_test.go +++ b/gateway/kube/examples/sharedstream/e2e_test.go @@ -28,7 +28,10 @@ import ( "k8s.io/utils/ptr" ) -// TestSharedHostRealAPI is manual evidence, included by task test-cluster. Set +// TestSharedHostRealAPI is manual evidence, included by task test-cluster. It needs +// EXCLUSIVE use of the cluster: activeConfigMapWatches reads a cluster-wide gauge with +// no namespace label, so any other client watching ConfigMaps is counted too. Run it +// with `go test -p 1`, never concurrently with the backend e2e suite. Set // KRM_SHARED_SUBSCRIBERS=200 for the opening/reconnect burst profile. No login // system or browser credentials are synthesized; the fixture supplies trusted sessions. func TestSharedHostRealAPI(t *testing.T) { From 20df2aad7e6d82e1a9f71e3e15cfd55351ea4190 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 11 Sep 2026 13:44:44 +0000 Subject: [PATCH 3/4] docs: record the 200-identity shared-host run as first-party evidence Adds docs/facts/shared-host-rehearsal.md for a manual run on k3s v1.36.2+k3s1: 200 independent identities synced in 109 ms and cost exactly one additional API-server ConfigMap watch, one withdrawn grant closed its stream in 27.3 s while 199 peers kept streaming, and the watch count returned to baseline. Measured from apiserver_longrunning_requests on the API server, not from the library's own counters. The file states what the run does not show: no proxy, no browsers, no resource limits, one sample rather than a distribution, and no statement of Kubernetes version support. Co-Authored-By: Claude Opus 5 (1M context) --- docs/facts/shared-host-rehearsal.md | 56 +++++++++++++++++++ .../0007-shared-stream-host-integration.md | 8 ++- gateway/kube/examples/sharedstream/README.md | 9 ++- 3 files changed, 68 insertions(+), 5 deletions(-) create mode 100644 docs/facts/shared-host-rehearsal.md diff --git a/docs/facts/shared-host-rehearsal.md b/docs/facts/shared-host-rehearsal.md new file mode 100644 index 0000000..10af8f9 --- /dev/null +++ b/docs/facts/shared-host-rehearsal.md @@ -0,0 +1,56 @@ +# Observed: the shared-stream host under 200 independent identities + +> **Recorded by hand from a manual run** — unlike [observed-v1.36.2+k3s1.md](observed-v1.36.2+k3s1.md), +> this file is not generated by `task cluster-facts`. It is a WITNESS to one run of one scenario on +> one disposable cluster. It is not a capacity guarantee, a latency distribution, or a statement of +> supported Kubernetes versions. + +## What was run + +| | | +|---|---| +| Commit | `ed5c4d0` (branch `feat/shared-stream-transport`, unreleased) | +| Scenario | `TestSharedHostRealAPI` in [`gateway/kube/examples/sharedstream`](../../gateway/kube/examples/sharedstream/) | +| Command | `KRM_SHARED_SUBSCRIBERS=200 go test -tags e2e -count=1 -p 1 -run TestSharedHostRealAPI ./examples/sharedstream/` | +| Cluster | k3s `v1.36.2+k3s1` via `task cluster-up`, Go `1.27.1` | +| Host config | `WriteTimeout=5s`, `ReauthorizationInterval=30s`, `ReauthorizationTimeout=5s` | +| Fixture client | `QPS=100`, `Burst=400` — declared fixture capacity, not a library default | + +## What it did + +| Observation | Result | +|---|---| +| 200 independent identities reached synced | 109.2 ms | +| API-server ConfigMap WATCH requests | baseline 26 → **27 while all 200 were streaming** | +| One RBAC grant withdrawn mid-stream | affected stream closed in 27.26 s; **199 peers retained** | +| After all disconnects | API watch count and host gauges returned to baseline | + +The **+1** is the claim worth keeping: 200 authenticated subscribers cost the API server exactly one +additional ConfigMap watch, measured from `apiserver_longrunning_requests` on the API server itself +rather than from the library's own counters. + +Revocation at 27.26 s sits inside Proposal 0006's 60-second reference budget, with a 30-second +recheck interval. One sample does not establish a bound: it does not include a stalled callback, a +blocked peer under write pressure, or an aligned timer cohort. + +## What this run does NOT show + +- **No proxy and no browsers.** Delivery went to an in-process `httptest` server over loopback. There + was no Traefik, no TLS termination hop, and no browser renderer or reconnect behavior. +- **No resource limits.** This ran in the devcontainer, not under a constrained CPU/memory budget, so + the numbers say nothing about behavior near saturation. +- **One run, not a distribution.** No percentiles, no sustained soak, no peak memory or CPU sample. +- **Service-account tokens, not logins.** The 200 identities are namespaced service accounts with + scoped RBAC. No OIDC provider or session system was exercised. +- **Not a version-support statement.** This is the repository's supported cluster version only. + +## Reproducing it + +`task test-cluster` runs this scenario at 2 subscribers as part of the manual gate. Set +`KRM_SHARED_SUBSCRIBERS` (2..200) for the burst profile. + +The scenario needs **exclusive use of the cluster**: `apiserver_longrunning_requests` is cluster-wide +and carries no namespace label, so any other ConfigMap watcher is counted in the same number. This is +why `task test-cluster` passes `-p 1`. Running the packages concurrently corrupts the reading — the +backend e2e suite's own real watches inflate it — and the failure looks like a watch leak that is not +there. diff --git a/docs/proposals/0007-shared-stream-host-integration.md b/docs/proposals/0007-shared-stream-host-integration.md index e6844f4..4ebc3b5 100644 --- a/docs/proposals/0007-shared-stream-host-integration.md +++ b/docs/proposals/0007-shared-stream-host-integration.md @@ -1,8 +1,10 @@ # Proposal 0007: Shared-stream transport correctness -**Status: implemented in the working tree; unreleased.** The transport, observation and example -increments below have landed. The real-cluster scenario in increment 3 is manual and has not been -run here, so no `docs/facts/` record exists yet. +**Status: implemented; unreleased.** The transport, observation and example increments below have +landed. The manual real-cluster scenario has been run at 2 and 200 identities and recorded in +[docs/facts/shared-host-rehearsal.md](../facts/shared-host-rehearsal.md); 200 subscribers cost +exactly one additional API-server watch. That is one run on one disposable cluster, without a proxy, +browsers or resource limits, and is not a capacity guarantee. ## Recommendation diff --git a/gateway/kube/examples/sharedstream/README.md b/gateway/kube/examples/sharedstream/README.md index 2574e1b..a134f6e 100644 --- a/gateway/kube/examples/sharedstream/README.md +++ b/gateway/kube/examples/sharedstream/README.md @@ -72,10 +72,15 @@ proof. Run the real-cluster test separately against a disposable cluster: ```bash # From gateway/kube, with the disposable cluster's KUBECONFIG selected: -go test -tags e2e -run '^TestSharedHostRealAPI$' -v -count=1 -timeout 4m ./examples/sharedstream -KRM_SHARED_SUBSCRIBERS=200 go test -tags e2e -run '^TestSharedHostRealAPI$' -v -count=1 -timeout 4m ./examples/sharedstream +go test -tags e2e -p 1 -run '^TestSharedHostRealAPI$' -v -count=1 -timeout 4m ./examples/sharedstream +KRM_SHARED_SUBSCRIBERS=200 go test -tags e2e -p 1 -run '^TestSharedHostRealAPI$' -v -count=1 -timeout 4m ./examples/sharedstream ``` +The scenario needs exclusive use of the cluster. `apiserver_longrunning_requests` is cluster-wide and +has no namespace label, so any other ConfigMap watcher lands in the same number; run it with `-p 1` +and never alongside the backend e2e suite. One recorded run at 200 identities is in +[docs/facts/shared-host-rehearsal.md](../../../../docs/facts/shared-host-rehearsal.md). + `task test-cluster` also includes the two-identity case. No new cluster CI workflow is required. The fixture creates isolated RBAC and independent service-account participant tokens, verifies participant writes, opening/reconnects, one revoked identity with continuing peers, balanced host From 8f6eaa60ea58ee2e32cac14536a54d84625101ba Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 11 Sep 2026 14:00:08 +0000 Subject: [PATCH 4/4] fix(kube): require verified HTTPS for the shared-host example cluster The example's Handler sends the service-account token and every participant token across the supplied rest.Config. A cleartext or Insecure configuration put both on the wire in the clear, and rest.IsConfigTransportTLS alone does not catch that: it only checks that the scheme is https and deliberately ignores Insecure. Reject both at construction, before any client exists. AnonymousClientConfig carries CAData, CAFile, ServerName and Insecure through to the participant client, so one check at construction covers both identities. The boundary test now runs against httptest.NewTLSServer with verification left on, trusting the server's own certificate rather than skipping the check, and asserts that cleartext and Insecure configurations are refused. The cluster fixture is unaffected: task cluster-kubeconfig already keeps verification on via --tls-server-name. Also state the working directory for the rehearsal command in docs/facts/shared-host-rehearsal.md; ./examples/sharedstream/ resolves relative to gateway/kube, which task test-cluster sets but a reader at the repository root does not. Both issues were raised by CodeRabbit on #35. Co-Authored-By: Claude Opus 5 (1M context) --- docs/facts/shared-host-rehearsal.md | 2 +- gateway/kube/CHANGELOG.md | 3 +- gateway/kube/examples/sharedstream/README.md | 8 +++-- gateway/kube/examples/sharedstream/handler.go | 12 ++++++-- .../examples/sharedstream/handler_test.go | 29 +++++++++++++++++-- 5 files changed, 45 insertions(+), 9 deletions(-) diff --git a/docs/facts/shared-host-rehearsal.md b/docs/facts/shared-host-rehearsal.md index 10af8f9..c1c26fe 100644 --- a/docs/facts/shared-host-rehearsal.md +++ b/docs/facts/shared-host-rehearsal.md @@ -11,7 +11,7 @@ |---|---| | Commit | `ed5c4d0` (branch `feat/shared-stream-transport`, unreleased) | | Scenario | `TestSharedHostRealAPI` in [`gateway/kube/examples/sharedstream`](../../gateway/kube/examples/sharedstream/) | -| Command | `KRM_SHARED_SUBSCRIBERS=200 go test -tags e2e -count=1 -p 1 -run TestSharedHostRealAPI ./examples/sharedstream/` | +| Command | from `gateway/kube`: `KRM_SHARED_SUBSCRIBERS=200 go test -tags e2e -count=1 -p 1 -run TestSharedHostRealAPI ./examples/sharedstream/` | | Cluster | k3s `v1.36.2+k3s1` via `task cluster-up`, Go `1.27.1` | | Host config | `WriteTimeout=5s`, `ReauthorizationInterval=30s`, `ReauthorizationTimeout=5s` | | Fixture client | `QPS=100`, `Burst=400` — declared fixture capacity, not a library default | diff --git a/gateway/kube/CHANGELOG.md b/gateway/kube/CHANGELOG.md index dbc9d15..ad74890 100644 --- a/gateway/kube/CHANGELOG.md +++ b/gateway/kube/CHANGELOG.md @@ -3,7 +3,8 @@ ## 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. Includes a + 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. diff --git a/gateway/kube/examples/sharedstream/README.md b/gateway/kube/examples/sharedstream/README.md index a134f6e..ce5b439 100644 --- a/gateway/kube/examples/sharedstream/README.md +++ b/gateway/kube/examples/sharedstream/README.md @@ -17,6 +17,10 @@ session expiry and token expiry. It must honor the request context. Browser-supp API-server headers are never used by the example. Keep the route behind the host's existing session protection; do not copy the integration test's fixture-session query routing into production. +`Handler` refuses a cluster configuration that is not HTTPS, and one with `Insecure` set: the +service token and every participant token cross that transport, and `rest.IsConfigTransportTLS` +alone would accept an unverified one. Supply a CA, not a skipped check. + The handler uses the server/TLS settings from the configured cluster for all three operations: - Participant SelfSubjectReview resolves username, groups, UID and extras at opening. Using the @@ -67,8 +71,8 @@ server writers. The gateway's socket tests separately verify stalled-peer behavi Normal `go test ./...` in `gateway/kube` compiles this example, checks subject ownership/API failures and cancellation, and proves that forged browser identity headers cannot change its identity or -scope. The tests use a real HTTP fake API for the credential boundary; they are not Kubernetes RBAC -proof. Run the real-cluster test separately against a disposable cluster: +scope. The tests use a verified-TLS fake API for the credential boundary, and assert that cleartext +and `Insecure` configurations are refused; they are not Kubernetes RBAC proof. Run the real-cluster test separately against a disposable cluster: ```bash # From gateway/kube, with the disposable cluster's KUBECONFIG selected: diff --git a/gateway/kube/examples/sharedstream/handler.go b/gateway/kube/examples/sharedstream/handler.go index ceba5ed..103d2c8 100644 --- a/gateway/kube/examples/sharedstream/handler.go +++ b/gateway/kube/examples/sharedstream/handler.go @@ -30,13 +30,21 @@ type resolvedSession struct { // Handler constructs one shared backend for one fixed ConfigMap on one cluster. // Call once at process startup. cluster supplies service credentials and server/TLS -// settings. Participant clients use only those server/TLS settings plus the trusted -// session token. The session callback must honor r.Context and own authentication. +// settings, and must use HTTPS with certificate verification: it carries both the +// service token and every participant token. Participant clients use only those +// server/TLS settings plus the trusted session token. The session callback must +// honor r.Context and own authentication. // Direct GET/PATCH routes must independently use participant credentials. func Handler(cluster *rest.Config, namespace, name string, sessionFor func(*http.Request) (Session, error), observer gateway.Observer) (http.Handler, error) { if cluster == nil || namespace == "" || name == "" || sessionFor == nil { return nil, fmt.Errorf("sharedstream: cluster, fixed scope and session resolver required") } + // Both the service token and every participant token cross this transport. IsConfigTransportTLS + // only proves the scheme is https; it ignores Insecure, which turns verification off and leaves + // the tokens readable to anything that can answer for the API server. + if !rest.IsConfigTransportTLS(*cluster) || cluster.Insecure { + return nil, fmt.Errorf("sharedstream: cluster must use HTTPS with certificate verification") + } service, err := kubernetes.NewForConfig(cluster) 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 e10e5c9..b34ad4d 100644 --- a/gateway/kube/examples/sharedstream/handler_test.go +++ b/gateway/kube/examples/sharedstream/handler_test.go @@ -3,6 +3,7 @@ package sharedstream import ( "context" "encoding/json" + "encoding/pem" "errors" "io" "net/http" @@ -97,7 +98,9 @@ func TestHostBoundaryUsesResolvedSubjectAndFixedScope(t *testing.T) { var subjects []authorizationv1.SubjectAccessReviewSpec var tokens []string var watches int - api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // A TLS server with verification left ON: the example refuses cleartext or unverified + // transports, and both tokens below would otherwise be readable on the wire. + api := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case strings.HasSuffix(r.URL.Path, "selfsubjectreviews"): @@ -134,9 +137,29 @@ func TestHostBoundaryUsesResolvedSubjectAndFixedScope(t *testing.T) { })) defer api.Close() counters := &Counters{} - h, err := Handler(&rest.Config{Host: api.URL, BearerToken: "service", ContentConfig: rest.ContentConfig{ContentType: "application/json"}}, "app", "coffee", func(*http.Request) (Session, error) { + verified := func() *rest.Config { + return &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(150 * time.Millisecond), TokenExpiry: time.Now().Add(time.Hour)}, nil - }, counters) + } + // Tokens must never reach a cleartext or unverified API server. + for _, refused := range []*rest.Config{ + {Host: strings.Replace(api.URL, "https://", "http://", 1), BearerToken: "service"}, + func() *rest.Config { c := verified(); c.Insecure, c.CAData = true, nil; return c }(), + } { + if _, err := Handler(refused, "app", "coffee", session, counters); err == nil { + t.Fatalf("accepted unverified transport: %s insecure=%t", refused.Host, refused.Insecure) + } + } + h, err := Handler(verified(), "app", "coffee", session, counters) if err != nil { t.Fatal(err) }