feat(gateway)!: share one stream configuration and move the test harness out of the gateway

README.md

@@ -151,18 +151,20 @@ Mount a scoped stream endpoint in an existing Go application:
 
 ```go
 mux.Handle("/resource-stream/v1", gateway.Handler(gateway.Options{
-	Principal:  func(r *http.Request) (gateway.Principal, error) { return userFromSession(r) },
-	Authorizer: authorizeScope,
-	Clients: func(_ context.Context, _ string, p gateway.Principal) (gateway.Backend, error) {
-		return kube.NewBackendForConfig(restConfigFor(p.(*User)))
-	},
+	Principal: func(r *http.Request) (gateway.Principal, error) { return userFromSession(r) },
 	Scopes: gateway.ScopePolicy{
 		Targets: []string{"production"},
 		Resources: []gateway.GroupResource{
 			{Resource: "configmaps", Scope: gateway.ResourceScopeNamespaced},
 		},
 	},
-	Projection: gateway.ProjectionFull,
+	StreamConfig: gateway.StreamConfig{
+		Authorizer: authorizeScope,
+		Clients: func(_ context.Context, _ string, p gateway.Principal) (gateway.Backend, error) {
+			return kube.NewBackendForConfig(restConfigFor(p.(*User)))
+		},
+		Projections: gateway.StaticProjection(gateway.ProjectionFull),
+	},
 }))
 ```
 

Taskfile.yml

@@ -50,6 +50,9 @@ tasks:
       - conformance/scopes.yaml
       - conformance/generate.sh
       - gateway/*.go
+      # The harness that replays each fixture into the goldens: change how a script is played, and the
+      # bytes it records can change with it.
+      - gateway/internal/conformance/*.go
       # The generator's own command line lives in this file, so it is a source too: changing WHICH
       # golden tests run, and then being told "up to date", is a trap I walked into writing this line.
       - Taskfile.yml

conformance/gen/sse/resourceversion-unorderable.sse

@@ -2,5 +2,5 @@
 
 data: {"seq":1,"type":"reset","target":"demo","scope":{"target":"demo","group":"wardle.example.com","version":"v1alpha1","resource":"flunders","namespace":"app"},"projection":"krm-full/v1"}
 
-data: {"seq":2,"type":"error","code":"INTERNAL","message":"upstream served a resourceVersion that cannot be ordered (opaque-alpha). Kubernetes 1.35+ requires every resourceVersion to be an orderable decimal, so this upstream is either older than 1.35 or an aggregated API server that does not conform. Set Gateway.Ordering = OrderingLenient to stream it anyway, without per-object monotonicity.","terminal":true}
+data: {"seq":2,"type":"error","code":"INTERNAL","message":"upstream served a resourceVersion that cannot be ordered (opaque-alpha). Kubernetes 1.35+ requires every resourceVersion to be an orderable decimal, so this upstream is either older than 1.35 or an aggregated API server that does not conform. Set StreamConfig.Ordering = OrderingLenient to stream it anyway, without per-object monotonicity.","terminal":true}
 

docs/adopting.md

@@ -29,29 +29,33 @@ func mount(mux *http.ServeMux, restConfigFor func(*User) *rest.Config) {
         Principal: func(r *http.Request) (gateway.Principal, error) {
             return userFromSession(r)
         },
-        // Deny before opening a watch; rechecked on every snapshot cycle.
-        Authorizer: gateway.AuthorizerFunc(func(ctx context.Context, p gateway.Principal, s gateway.Scope) error {
-            user := p.(*User)
-            if s.Target != user.Target || s.Namespace != user.Namespace {
-                return gateway.Forbidden("scope is not available to this user")
-            }
-            return nil
-        }),
-        // A backend acting as this user.
-        Clients: func(_ context.Context, _ string, p gateway.Principal) (gateway.Backend, error) {
-            return kube.NewBackendForConfig(restConfigFor(p.(*User)))
-        },
         Scopes: gateway.ScopePolicy{
             Targets: []string{"production"},
             Resources: []gateway.GroupResource{
                 {Group: "", Resource: "configmaps", Scope: gateway.ResourceScopeNamespaced},
                 {Group: "apps", Resource: "deployments", Scope: gateway.ResourceScopeNamespaced},
             },
         },
-        Projection:   gateway.ProjectionFull,
-        WriteTimeout: 10 * time.Second, // bound each write to a browser that stops reading
-        Diagnostics: func(d gateway.Diagnostic) {
-            slog.Warn("stream error", "code", d.Code, "resource", d.Scope.Resource, "err", redact(d.Err))
+        // Settings a stream needs however it is served; a Gateway embeds the same StreamConfig.
+        StreamConfig: gateway.StreamConfig{
+            // Deny before opening a watch; rechecked on every snapshot cycle.
+            Authorizer: gateway.AuthorizerFunc(func(ctx context.Context, p gateway.Principal, s gateway.Scope) error {
+                user := p.(*User)
+                if s.Target != user.Target || s.Namespace != user.Namespace {
+                    return gateway.Forbidden("scope is not available to this user")
+                }
+                return nil
+            }),
+            // A backend acting as this user.
+            Clients: func(_ context.Context, _ string, p gateway.Principal) (gateway.Backend, error) {
+                return kube.NewBackendForConfig(restConfigFor(p.(*User)))
+            },
+            // One view for everyone. Nil means the same; a ProjectionPolicy can choose per caller.
+            Projections:  gateway.StaticProjection(gateway.ProjectionFull),
+            WriteTimeout: 10 * time.Second, // bound each write to a browser that stops reading
+            Diagnostics: func(d gateway.Diagnostic) {
+                slog.Warn("stream error", "code", d.Code, "resource", d.Scope.Resource, "err", redact(d.Err))
+            },
         },
     })
     mux.Handle("/resource-stream/v1", endWithSession(stream))
@@ -146,7 +150,7 @@ Plan with the [revocation budget](auth.md#revocation-budget).
 the scope:
 
 ```go
-shared := gateway.NewSharedBackendWithOptions(serviceAccountBackend, gateway.SharedOptions{
+shared := gateway.NewSharedBackend(serviceAccountBackend, gateway.SharedOptions{
     QueueDepth: 512,
     Observer:   metrics,
 })
@@ -167,7 +171,7 @@ monitoring and limits.
 ### Errors and diagnostics
 
 A `StreamError` you return keeps its `Message`; its `Cause` never reaches the wire. Any other error
-reaches the browser as `INTERNAL` with the message `internal error`. `Options.Diagnostics` receives
+reaches the browser as `INTERNAL` with the message `internal error`. `StreamConfig.Diagnostics` receives
 the raw error with its principal and scope; log what your policy allows.
 
 `Principal` may return `gateway.Unauthenticated` when signing in again will help, or
@@ -203,7 +207,7 @@ your own tests.
 A `rest.Config.Host` may include a path prefix, such as a kcp workspace URL
 (`https://kcp.example/clusters/root:org:ws`); the dynamic client preserves it.
 `kube.NewBackendForConfig(cfg)` includes the endpoint in upstream errors, which reach
-`Options.Diagnostics` and never the browser. To use `kube.NewBackend(dynamicClient)` instead, build
+`StreamConfig.Diagnostics` and never the browser. To use `kube.NewBackend(dynamicClient)` instead, build
 the client on `kube.HTTPClientFor(cfg)` (`dynamic.NewForConfigAndClient(cfg, httpClient)`), which
 refuses redirects so a user's token is not carried along.
 

docs/auth.md

@@ -84,10 +84,9 @@ checks; zero timeout uses 10 seconds. Timed checks do not call `Clients`.
 A timed check shares the subscriber's delivery gate: while it runs, that subscriber receives no
 objects, and it waits for the delivery in progress. A write to a browser that has stopped reading
 blocks once the buffers fill, so HTTP serving requires a positive `WriteTimeout` with timed checks.
-`Handler` panics at construction without one, and `ServeStream` and `ServeStreamProjection` panic
-before writing. Test that your mounted middleware supports flushing and write deadlines with the
+`Handler` panics at construction without one, and `ServeStream` panics before writing. Test that your mounted middleware supports flushing and write deadlines with the
 [capability-check recipe](../gateway/kube/examples/sharedstream/README.md#middleware-capability-test).
-`Stream` and `StreamProjection` allow timed checks with any sink; the host bounds that sink's I/O.
+`Stream` allows timed checks with any sink; the host bounds that sink's I/O and honors cancellation.
 
 Each check costs the host's authorizer a call. With `SubjectAccessReviewAuthorizer`, 200 subscribers
 on a 30-second interval add about 13 reviews per second (list and watch each), plus opening and

docs/operations.md

@@ -12,7 +12,7 @@ metrics := gateway.ObserverFunc(func(o gateway.Observation) {
 Do not block in `Observe`; it runs on the stream or shared-watch goroutine. Never label a metric with
 an object name, UID, principal, patch contents, scope value or error message.
 
-Error text goes to `Options.Diagnostics` instead, because it can name internal addresses and URLs; the
+Error text goes to `StreamConfig.Diagnostics` instead, because it can name internal addresses and URLs; the
 browser receives only the protocol code and a message chosen for it. Redact before logging.
 
 ## Signals to alert on
@@ -55,22 +55,22 @@ browser receives only the protocol code and a message chosen for it. Redact befo
 
 | Control | Default | Use |
 |---|---:|---|
-| `gateway.Options.WriteTimeout` | 0 (no deadline) | set a positive per-operation write-plus-flush budget for bounded HTTP delivery; required with `ReauthorizationInterval` |
-| `gateway.Options.ReauthorizationInterval` | 0 (cycle checks only) | recheck each subscriber on quiet streams; see the [revocation budget](auth.md#revocation-budget) |
-| `gateway.Options.ReauthorizationTimeout` | 10 seconds | bound each periodic check's callbacks once it holds the delivery gate |
-| `gateway.Options.HeartbeatInterval` | 20 seconds | set below the shortest proxy idle timeout |
+| `StreamConfig.WriteTimeout` | 0 (no deadline) | set a positive per-operation write-plus-flush budget for bounded HTTP delivery; required with `ReauthorizationInterval` |
+| `StreamConfig.ReauthorizationInterval` | 0 (cycle checks only) | recheck each subscriber on quiet streams; see the [revocation budget](auth.md#revocation-budget) |
+| `StreamConfig.ReauthorizationTimeout` | 10 seconds | bound each periodic check's callbacks once it holds the delivery gate |
+| `StreamConfig.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 |
 | `GroupResource.AllowAllNamespaces` | false | make all-namespaces access an explicit reviewable policy decision |
-| `Gateway.Ordering` | strict | keep strict on supported Kubernetes; use lenient only for known aggregated APIs |
-| `gateway.Options.Diagnostics` | nil (discarded) | receive the raw error behind each error event; redact before logging |
+| `StreamConfig.Ordering` | strict | keep strict on supported Kubernetes; use lenient only for known aggregated APIs |
+| `StreamConfig.Diagnostics` | nil (discarded) | receive the raw error behind each error event; redact before logging |
 
 The gateway sets no snapshot object or byte limit: what is safe depends on the product. 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
+`stream_opened`/`stream_closed` count `Stream` 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. A resnapshot can

gateway/README.md

@@ -7,35 +7,43 @@ SubjectAccessReview authorizer.
 
 ```go
 handler := gateway.Handler(gateway.Options{
-	Principal:  principalFromSession, // who is calling
-	Authorizer: authorizeScope,       // may they open this scope?
-	Clients: func(_ context.Context, target string, p gateway.Principal) (gateway.Backend, error) {
-		return kube.NewBackendForConfig(restConfigFor(target, p)) // acting as the caller
-	},
+	Principal: principalFromSession, // who is calling
 	Scopes: gateway.ScopePolicy{ // the allowlist; the zero value denies everything
 		Targets: []string{"production"},
 		Resources: []gateway.GroupResource{
 			{Group: "", Resource: "configmaps", Scope: gateway.ResourceScopeNamespaced},
 		},
 	},
-	Projection:   gateway.ProjectionFull,
-	WriteTimeout: 10 * time.Second,
+	StreamConfig: gateway.StreamConfig{
+		Authorizer: authorizeScope, // may they open this scope?
+		Clients: func(_ context.Context, target string, p gateway.Principal) (gateway.Backend, error) {
+			return kube.NewBackendForConfig(restConfigFor(target, p)) // acting as the caller
+		},
+		Projections:  gateway.StaticProjection(gateway.ProjectionFull),
+		WriteTimeout: 10 * time.Second,
+	},
 })
 ```
 
 `Handler` panics at construction on a missing seam or an unsafe option. The gateway never accepts an
 API-server URL or a credential from a browser request.
 
+`StreamConfig` holds the settings a stream needs however it is served, with one name each. `Options`
+adds the HTTP-only `Principal` and `Scopes`; a host that routes and authorizes its own requests embeds
+the same `StreamConfig` in a `Gateway` and calls `ServeStream`, or `Stream` with its own `Sink`. Both
+take the projection name the caller requested; empty asks the `Projections` policy for its default.
+Nil `Projections` grants only `ProjectionFull`.
+
 ## Stream behavior
 
 Every snapshot cycle emits `reset`, zero or more `added` events, then `synced`; live updates are
 complete-object replacements. Lost upstream continuity emits `RESYNC_REQUIRED` and starts a new cycle
 on the same connection. Any other error ends the connection: a non-terminal one such as
 `UPSTREAM_UNAVAILABLE` carries `retryAfterMs` and the client retries; a terminal one is the final
-event. Unexpected error text goes only to `Options.Diagnostics`.
+event. Unexpected error text goes only to `StreamConfig.Diagnostics`.
 
 The gateway absorbs bookmarks, relists, 410 responses, partial metadata objects and ambiguous
-deletion tombstones. `Gateway.Ordering` defaults to strict decimal `resourceVersion` ordering for
+deletion tombstones. `StreamConfig.Ordering` defaults to strict decimal `resourceVersion` ordering for
 Kubernetes 1.35+, preserving per-object monotonicity within a cycle; use `OrderingLenient` only for an
 upstream whose versions cannot be ordered. The normative details are in [`spec/v1.md`](../spec/v1.md).
 

gateway/auth_test.go

@@ -66,18 +66,20 @@ func TestRevokedAccessTerminatesAnOpenStream(t *testing.T) {
 
 	sink := &authSink{}
 	g := &Gateway{
-		// Entitlement changes UNDER the stream: the first cycle is authorized, and by the time the
-		// gateway asks again — which, before this was fixed, it never did — the answer has flipped.
-		Auth: AuthorizerFunc(func(context.Context, Principal, Scope) error {
-			if asked.Add(1) == 1 {
-				return nil
-			}
-			return Forbidden("your access to this scope was revoked")
-		}),
-		Clients: func(context.Context, string, Principal) (Backend, error) { return &closingBackend{}, nil },
+		StreamConfig: StreamConfig{
+			// Entitlement changes UNDER the stream: the first cycle is authorized, and by the time the
+			// gateway asks again — which, before this was fixed, it never did — the answer has flipped.
+			Authorizer: AuthorizerFunc(func(context.Context, Principal, Scope) error {
+				if asked.Add(1) == 1 {
+					return nil
+				}
+				return Forbidden("your access to this scope was revoked")
+			}),
+			Clients: func(context.Context, string, Principal) (Backend, error) { return &closingBackend{}, nil },
+		},
 	}
 
-	_ = g.Stream(t.Context(), "alice", Scope{Version: "v1", Resource: "configmaps"}, sink)
+	_ = g.Stream(t.Context(), "alice", Scope{Version: "v1", Resource: "configmaps"}, "", sink)
 
 	if asked.Load() < 2 {
 		t.Fatal("the Authorizer was consulted once and never again: a revoked user's open stream " +
@@ -102,20 +104,22 @@ func TestTheClientIsResolvedOnEveryCycle(t *testing.T) {
 
 	sink := &authSink{}
 	g := &Gateway{
-		Auth: AllowAll{},
-		Clients: func(context.Context, string, Principal) (Backend, error) {
-			// Three cycles is enough to show it is per-cycle and not once-ever.
-			if resolved.Add(1) >= 3 {
-				return nil, Forbidden("the credential could not be refreshed")
-			}
-			return backend, nil
-		},
 		// Each cycle stands in for a long-lived one: without this, instant cycles are an upstream
 		// ending every watch early, which the gateway stops reopening after the second.
 		now: steppingClock(2 * time.Second),
+		StreamConfig: StreamConfig{
+			Authorizer: AllowAll{},
+			Clients: func(context.Context, string, Principal) (Backend, error) {
+				// Three cycles is enough to show it is per-cycle and not once-ever.
+				if resolved.Add(1) >= 3 {
+					return nil, Forbidden("the credential could not be refreshed")
+				}
+				return backend, nil
+			},
+		},
 	}
 
-	_ = g.Stream(t.Context(), "alice", Scope{Version: "v1", Resource: "configmaps"}, sink)
+	_ = g.Stream(t.Context(), "alice", Scope{Version: "v1", Resource: "configmaps"}, "", sink)
 
 	if got := resolved.Load(); got < 3 {
 		t.Errorf("ClientFor was called %d times across several cycles — a host has nowhere to refresh "+
@@ -135,13 +139,15 @@ func TestClientResolutionReceivesTheStreamContext(t *testing.T) {
 	var received any
 
 	g := &Gateway{
-		Auth: AllowAll{},
-		Clients: func(ctx context.Context, _ string, _ Principal) (Backend, error) {
-			received = ctx.Value(contextKey{})
-			return nil, Forbidden("stop after checking context")
+		StreamConfig: StreamConfig{
+			Authorizer: AllowAll{},
+			Clients: func(ctx context.Context, _ string, _ Principal) (Backend, error) {
+				received = ctx.Value(contextKey{})
+				return nil, Forbidden("stop after checking context")
+			},
 		},
 	}
-	_ = g.Stream(ctx, "alice", Scope{Version: "v1", Resource: "configmaps"}, &authSink{})
+	_ = g.Stream(ctx, "alice", Scope{Version: "v1", Resource: "configmaps"}, "", &authSink{})
 	if received != "request-value" {
 		t.Errorf("ClientFor context value = %v, want request context value", received)
 	}
@@ -153,11 +159,13 @@ func TestDenialOpensNoWatchAtAll(t *testing.T) {
 	backend := &closingBackend{}
 	sink := &authSink{}
 	g := &Gateway{
-		Auth:    AuthorizerFunc(func(context.Context, Principal, Scope) error { return Forbidden("no") }),
-		Clients: func(context.Context, string, Principal) (Backend, error) { return backend, nil },
+		StreamConfig: StreamConfig{
+			Authorizer: AuthorizerFunc(func(context.Context, Principal, Scope) error { return Forbidden("no") }),
+			Clients:    func(context.Context, string, Principal) (Backend, error) { return backend, nil },
+		},
 	}
 
-	_ = g.Stream(t.Context(), "mallory", Scope{Version: "v1", Resource: "secrets"}, sink)
+	_ = g.Stream(t.Context(), "mallory", Scope{Version: "v1", Resource: "secrets"}, "", sink)
 
 	if backend.opened.Load() != 0 {
 		t.Error("a watch was opened for a caller who was refused — the object's existence has leaked")