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/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/facts/shared-host-rehearsal.md b/docs/facts/shared-host-rehearsal.md new file mode 100644 index 0000000..c1c26fe --- /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 | 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 | + +## 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/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..4ebc3b5 --- /dev/null +++ b/docs/proposals/0007-shared-stream-host-integration.md @@ -0,0 +1,225 @@ +# Proposal 0007: Shared-stream transport correctness + +**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 + +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..ad74890 100644 --- a/gateway/kube/CHANGELOG.md +++ b/gateway/kube/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## Unreleased + +- Add a tested, copyable shared ConfigMap host example with local full-subject resolution, + session expiry, service-account SARs, bounded SSE delivery and lifecycle counters. It requires a + verified-HTTPS cluster configuration, since it carries service and participant tokens. Includes a + manually invoked independent-identity Kubernetes fixture; no new public identity helper. + + ## [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..ce5b439 --- /dev/null +++ b/gateway/kube/examples/sharedstream/README.md @@ -0,0 +1,111 @@ +# 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. + +`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 + 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 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: +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 +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..350aec8 --- /dev/null +++ b/gateway/kube/examples/sharedstream/e2e_test.go @@ -0,0 +1,308 @@ +//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. 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) { + 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..103d2c8 --- /dev/null +++ b/gateway/kube/examples/sharedstream/handler.go @@ -0,0 +1,117 @@ +// 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, 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 + } + 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..b34ad4d --- /dev/null +++ b/gateway/kube/examples/sharedstream/handler_test.go @@ -0,0 +1,224 @@ +package sharedstream + +import ( + "context" + "encoding/json" + "encoding/pem" + "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 + // 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"): + 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{} + 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 + } + // 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) + } + 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) } }