From 007b6dc8712306c5b71e90ac7f6f38dfb2e0b7c3 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 11 Sep 2026 07:57:48 +0000 Subject: [PATCH 1/9] feat: add managed recovery and safe live-editor integration --- README.md | 2 +- docs/adopting.md | 16 +- docs/auth.md | 30 +++- docs/proposals/0004-views-and-bytes.md | 9 +- docs/saving.md | 82 ++++----- docs/vue.md | 43 +++++ docs/why-a-gateway.md | 2 +- examples/README.md | 3 + examples/conditional-save/README.md | 43 +++++ examples/conditional-save/editor.ts | 45 +++++ .../vanilla-browser/tests/live-krm.spec.ts | 84 ++++++++- gateway/README.md | 8 +- gateway/handler.go | 23 ++- gateway/kube/authz.go | 12 +- gateway/kube/e2e_test.go | 43 +++++ .../kube/examples/conditionalsave/handler.go | 137 ++++++++++++++ .../examples/conditionalsave/handler_test.go | 62 +++++++ gateway/reauthorize.go | 79 +++++++++ gateway/reauthorize_test.go | 167 ++++++++++++++++++ gateway/shared.go | 4 +- gateway/stream.go | 14 +- packages/krm-stream/README.md | 46 ++++- packages/krm-stream/src/connection.ts | 132 ++++++++++++++ packages/krm-stream/src/index.ts | 5 +- packages/krm-stream/src/sse.ts | 54 ++++-- packages/krm-stream/src/store.ts | 55 +++++- packages/krm-stream/test/connection.test.ts | 165 +++++++++++++++++ packages/krm-stream/test/saving.test.ts | 120 +++++++++++++ packages/krm-stream/tsconfig.test.json | 4 +- 29 files changed, 1386 insertions(+), 103 deletions(-) create mode 100644 docs/vue.md create mode 100644 examples/conditional-save/README.md create mode 100644 examples/conditional-save/editor.ts create mode 100644 gateway/kube/examples/conditionalsave/handler.go create mode 100644 gateway/kube/examples/conditionalsave/handler_test.go create mode 100644 gateway/reauthorize.go create mode 100644 gateway/reauthorize_test.go create mode 100644 packages/krm-stream/src/connection.ts create mode 100644 packages/krm-stream/test/connection.test.ts create mode 100644 packages/krm-stream/test/saving.test.ts diff --git a/README.md b/README.md index 89bd752..6664453 100644 --- a/README.md +++ b/README.md @@ -164,7 +164,7 @@ patch through its own save endpoint, which is the one place a write can happen. | Package | Purpose | |---|---| | `github.com/ConfigButler/krm-stream/gateway` | Dependency-free Go stream gateway and SSE handler. | -| `github.com/ConfigButler/krm-stream/gateway/kube` | Optional `client-go` backend and SSAR authorizer. | +| `github.com/ConfigButler/krm-stream/gateway/kube` | Optional `client-go` backend and SubjectAccessReview authorizer. | | `@configbutler/krm-stream` | Official dependency-free ESM client store and transports. | | `krm-stream@0.1.0` | Deprecated, frozen compatibility name claim. Use the scoped package instead. | | [`spec/v1.md`](spec/v1.md) | Normative protocol contract. | diff --git a/docs/adopting.md b/docs/adopting.md index f2fb1ac..befe713 100644 --- a/docs/adopting.md +++ b/docs/adopting.md @@ -91,8 +91,8 @@ func mount(mux *http.ServeMux, dynamicClientFor func(*User) dynamic.Interface) { } ``` -The browser uses `connectWithEventSource` for this route. Its same-origin session cookie is the only -credential EventSource can carry. +The browser uses `connectManagedResourceStream` for this route. Fetch sends its same-origin session +cookie and the managed connection provides bounded recovery after network failures and sequence gaps. ## 3. Browser client @@ -100,7 +100,7 @@ credential EventSource can carry. import { defaultPolicy, LiveResourceStore, - connectWithEventSource, + connectManagedResourceStream, resourceStreamURL, withOpenAPIKeyedLists, } from "@configbutler/krm-stream"; @@ -114,13 +114,13 @@ const url = resourceStreamURL("/resource-stream/v1", { projection: "krm-full/v1", }); -connectWithEventSource(url, store, { - onGap: () => location.reload(), // reconnect for a new snapshot in an SPA-specific way +const connection = connectManagedResourceStream(url, store, { + onStateChange: state => renderConnection(state.status), // gaps recover with a fresh snapshot }); store.subscribe(() => render(store)); ``` -For a bearer-token client, use `connectResourceStream(url, store, { headers: { Authorization: ... } })`. +For a bearer-token client, use `connectManagedResourceStream(url, store, { headers: { Authorization: ... } })`. That is useful for a non-browser client or an intentionally token-bearing browser application; the same-origin cookie route is the safer browser default. @@ -138,7 +138,7 @@ stays safely atomic. ## 4. Share watches only with Kubernetes-backed authorization `SharedBackend` saves upstream watches but runs as one service identity. Pair it with -`kube.SSARAuthorizer` so Kubernetes still decides whether each caller may list and watch the scope. +`kube.SubjectAccessReviewAuthorizer` so Kubernetes still decides whether each caller may list and watch the scope. ```go shared := gateway.NewSharedBackendWithOptions(serviceAccountBackend, gateway.SharedOptions{ @@ -146,7 +146,7 @@ shared := gateway.NewSharedBackendWithOptions(serviceAccountBackend, gateway.Sha Observer: metrics, }) -options.Authorizer = kube.SSARAuthorizer(clientset, subjectFromUser) +options.Authorizer = kube.SubjectAccessReviewAuthorizer(clientset, subjectFromUser) options.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend, error) { return shared, nil } ``` diff --git a/docs/auth.md b/docs/auth.md index 2a706e5..da2bd00 100644 --- a/docs/auth.md +++ b/docs/auth.md @@ -103,10 +103,25 @@ too: - **`ClientFor` is your refresh point.** It is called again each cycle, so you can hand back a client bearing a *fresh* token. -**The gap:** a perfectly quiet stream may not cycle for a long time, so revocation is -noticed at the *next cycle*, not instantly. The credential half of that is solved properly on your -side of the seam — give the client a **refreshing token source** (Dex issues a refresh token; the -standard `oauth2.TokenSource` wraps it), and it never hands us a dead token in the first place. +Set `ReauthorizationInterval` to bound how long a quiet stream runs without checking entitlement: + +```go +options.ReauthorizationInterval = 30 * time.Second +options.ReauthorizationTimeout = 5 * time.Second +``` + +Timed checks are per subscriber and recheck both `Authorizer` and the projection policy. During a +check that subscriber's object delivery pauses. Denial, timeout or policy failure terminates only +that stream; other subscribers and the shared upstream continue. A changed projection terminates +the old stream so it cannot keep disclosing its previous view. Zero interval preserves cycle-only +checks; zero timeout uses 10 seconds. The bound assumes host callbacks honor context cancellation +and sinks do not block indefinitely. The check uses the principal captured at stream open: resolve +current session/account validity inside the host authorizer if those can change independently of RBAC. + +For 200 subscribers, a 30-second interval adds roughly 13 SubjectAccessReviews/second (list and watch +per subscriber), plus opening/cycle checks. Choose an interval and timeout for your revocation budget +and API-server capacity; checks are not cached across identities. `ClientFor` still runs per snapshot +cycle so the host can return a client backed by refreshing credentials. ## Two things that are easy to confuse @@ -120,11 +135,11 @@ scope, so it opens it **once**, so it opens it as **one identity** — your serv moment your `Authorizer` stops being defence in depth and becomes *the only thing* between a caller and the objects. That is why it is opt-in, and why it is not the default. -If you turn it on, use **`kube.SSARAuthorizer`**, and Kubernetes is the boundary again: +If you turn it on, use **`kube.SubjectAccessReviewAuthorizer`**, and Kubernetes is the boundary again: ```go shared := gateway.NewSharedBackend(serviceAccountBackend) // one watch, one identity… -opts.Authorizer = kube.SSARAuthorizer(clientset, subjectOf) // …but RBAC still decides +opts.Authorizer = kube.SubjectAccessReviewAuthorizer(clientset, subjectOf) // …but RBAC still decides opts.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend, error) { return shared, nil } ``` @@ -145,7 +160,8 @@ Three things it does that are easy to get accidentally permissive, all tested: It needs your server's service account to hold `create` on `subjectaccessreviews` (the standard `system:auth-delegator` role). It does **not** need impersonate rights: it asks a question *about* a user, it does not act *as* one. And because the gateway re-authorizes every snapshot cycle, this is -also how a revocation reaches a stream that is already open. +also how a revocation reaches a stream that is already open. Timed checks bound quiet-stream revocation. +The old `SSARAuthorizer` name remains a deprecated alias; it never created SelfSubjectAccessReview. ## What this library never does diff --git a/docs/proposals/0004-views-and-bytes.md b/docs/proposals/0004-views-and-bytes.md index c6a5f10..97ca93c 100644 --- a/docs/proposals/0004-views-and-bytes.md +++ b/docs/proposals/0004-views-and-bytes.md @@ -104,16 +104,17 @@ actually emitted, including `reset`, `synced`, errors, and deletes. It does not cycle. A new HTTP connection begins at one. The TypeScript transport checks every sequence number. A missing, repeated, malformed, or out-of-order -number closes the transport and reports a gap; the host reconnects for a fresh snapshot. `seq` is not +number closes the transport and reports a gap; the managed connector retries for a fresh snapshot. `seq` is not an SSE `id` and does not provide replay or resume semantics. ## Consequences - `krm-spec/v1` makes a status-blind editor cheap under controller churn without weakening the complete-object invariant for the fields it receives. -- A suppressed update may leave the consumer's `metadata.resourceVersion` stale. Consumers must not - use that opaque value as a save precondition. Saves remain narrow merge patches built from local - edits, while the client-side three-way merge reports visible conflicts live. +- A suppressed update may leave the consumer's `metadata.resourceVersion` stale. It remains a safe + save precondition: the API server rejects a stale write with 409. Capture it with the patch, + reconcile before retrying, and never substitute a new version onto an old patch. Narrow patches + limit write scope; they do not prevent lost updates. See [saving](../saving.md). - Shared upstream watches remain safe: projection, redaction revision, suppression digest, and sequence state are per consumer stream, after upstream fan-out. diff --git a/docs/saving.md b/docs/saving.md index 00b3185..7ebc287 100644 --- a/docs/saving.md +++ b/docs/saving.md @@ -4,37 +4,40 @@ krm-stream is a read library. Your application owns its HTTP save endpoint, audi Kubernetes client. The client store produces a narrow RFC 7386 JSON merge patch; use `gateway.ValidateMergePatch` immediately before sending it to Kubernetes. -```go -func (s *server) saveConfigMap(w http.ResponseWriter, r *http.Request) { - user := userFromSession(r) - scope := authorizedScope(user, r) - patch := readRequestBody(r) - - // Read the current object using the same identity and effective projection decision as the stream. - object := s.currentObject(r.Context(), user, scope) - if err := gateway.ValidateMergePatch(gateway.ProjectionFull, object, patch); err != nil { - var violation *gateway.PatchViolation - if errors.As(err, &violation) { - http.Error(w, violation.Error(), http.StatusBadRequest) - return - } - http.Error(w, "invalid merge patch", http.StatusBadRequest) - return - } +Use the complete [conditional-save example](../examples/conditional-save/README.md): it includes a +compilable host endpoint, client reconciliation, race tests, and a real-cluster 409 test. - // Your app performs this ordinary Kubernetes PATCH as the signed-in user. - _, err := s.dynamicFor(user).Resource(configMaps).Namespace(scope.Namespace). - Patch(r.Context(), scope.Name, types.MergePatchType, patch, metav1.PatchOptions{}) - if err != nil { - http.Error(w, "save failed", http.StatusBadGateway) - return - } +```ts +const intent = store.captureSave(uid); +// intent = { uid, resourceVersion, patch }, detached and captured together before any await. +if (intent) await hostSave(intent); +``` - // 204, and NOT the object Kubernetes just handed back. See below. - w.WriteHeader(http.StatusNoContent) -} +The host validates the patch and adds `metadata.uid` and `metadata.resourceVersion` from that intent +before sending a Kubernetes merge PATCH. It must not substitute the version from a newer GET. +Kubernetes performs the version check atomically with the write. Propagate a real Kubernetes 409 +as HTTP 409; do not turn all upstream errors into 502. + +A narrow patch limits which fields are written; it does **not** protect against concurrent edits. +JSON merge patch replaces arrays in full, including arrays the client reconciles by keyed items. +See Kubernetes' [conditional update guidance](https://kubernetes.io/docs/reference/using-api/api-concepts/#updates-to-existing-resources). + +On 409, keep the draft and reconcile a fresh, complete **projected** GET. Capture the response guard +before starting that GET: + +```ts +const reconcile = store.captureReconciliation(uid); +const { object, redacted } = await hostRead(); +reconcile(object, { redacted }); +// false means a newer server event/response won, or the UID disappeared/changed. Do not force it. ``` +Local edits made during the request survive reconciliation. Render the current draft and conflicts; +let the user review before capturing another save intent. Serialize saves per editor. A host GET must +be a most-recent Kubernetes read with no intermediary cache and the same projection as the stream. +Do not apply a response for a replacement UID to the old editor. Deleted-object drafts are removed by +the store; archive them outside the store if the application needs recovery after deletion. + ## What the guard rejects - a value declared in `redacted`, including deletion of a parent map such as `data: null`; @@ -47,9 +50,10 @@ It does not grant write permission, choose a projection, fetch the object, issue optimistic concurrency. Those stay with the host. Do not use whole-object `PUT`: projected objects are intentionally incomplete, and a `PUT` can delete fields the browser never saw. -`metadata.resourceVersion` may be stale when `krm-spec/v1` suppresses invisible status churn. Do not -use the streamed value as a write precondition. The client-side three-way merge surfaces conflicts in -the fields the user can see; send only the user's explicit merge-patch changes. +`metadata.resourceVersion` may be stale when a projection suppresses invisible changes. It remains +safe as a write precondition: a stale version causes a 409 instead of a lost update. Reconcile a fresh +projected read and capture a new intent after review. Invisible churn may cause extra conflicts; +removing concurrency protection to avoid those conflicts is unsafe. ## Answer 204 and let the watch echo it @@ -65,17 +69,15 @@ clear and nothing to adopt. The echo settles it. ## If you must answer with the object -`store.adoptSaved(object)` is for a host that already holds a **projected** object, such as one doing -its own optimistic update. Project it first: - -```go -projected, redacted := gateway.Project(gateway.ProjectionFull, result) -_ = redacted // the paths withheld; the client keeps the redactions it already has -writeJSON(w, projected) -``` +Prefer 204. If a host returns an object, capture `store.captureReconciliation(uid)` **before** the +save request and apply its projected response through that guard. This prevents a delayed response +from overwriting a newer watch event, and preserves local edits made while saving. -`gateway.Project` applies the same projection the stream applies. Never hand `adoptSaved` an object -straight from the Kubernetes client. +`store.adoptSaved(object)` remains available for synchronous adoption and newly created objects. It +is unguarded and must not receive delayed responses that can race the watch. Never adopt a raw +Kubernetes response: project it first and provide the correct redaction metadata. Hosts exposing +redacted resources must keep read and stream redaction revisions consistent; the ConfigMap example +has no withheld values. ## Creating and deleting whole objects diff --git a/docs/vue.md b/docs/vue.md new file mode 100644 index 0000000..b3ada01 --- /dev/null +++ b/docs/vue.md @@ -0,0 +1,43 @@ +# Vue integration + +A copyable Vue 3 composable covers the adapter: reactive snapshots for drafts, changes, conflicts, +redactions and connection state, plus automatic subscription cleanup. Vue remains a host dependency; +the core package and its browser bundle have no framework dependencies. + +```ts +import { shallowRef, onScopeDispose } from "vue"; +import type { LiveResourceStore, ManagedStreamHandle } from "@configbutler/krm-stream"; + +// Call in setup() or an active effect scope. One fixed UID per instance. +export function useLiveResource( + store: LiveResourceStore, + uid: string, + connection: ManagedStreamHandle, +) { + const read = () => store.ids().includes(uid) ? { + draft: store.draft(uid), + changes: store.changes(uid), + conflicts: store.conflicts(uid), + redactions: store.redactions(uid), + } : null; + const resource = shallowRef(read()); + const state = shallowRef(connection.state); + const stopStore = store.subscribe(() => { resource.value = read(); }); + const stopConnection = connection.subscribe(next => { state.value = next; }); + onScopeDispose(() => { stopStore(); stopConnection(); }); + return { resource, state }; +} +``` + +Use `resource.value?.draft` in script and `resource?.draft` in a template. A missing/deleted UID gives +`null`. For a changing selection, remount a keyed editor or create a new effect scope. Use store +methods such as `setValue`, `removeKey` and conflict resolution methods for edits: do not bind +`v-model` directly to the draft snapshot. Store reads are detached copies, and editing them bypasses +policy checks and change notifications. + +The connection belongs to its creator. A parent sharing one connection across many editors closes +it when the parent unmounts; an editor that creates its own connection can register +`onScopeDispose(() => connection.close())`. Removing one editor's subscriptions must not close a +connection used by other editors. Capture and submit saves with the +[conditional-save example](../examples/conditional-save/README.md); expose request errors and saving +state from that host-owned controller. Enable Save while `state.status === "live"`. diff --git a/docs/why-a-gateway.md b/docs/why-a-gateway.md index 5d598f2..fdc8eeb 100644 --- a/docs/why-a-gateway.md +++ b/docs/why-a-gateway.md @@ -57,7 +57,7 @@ one identity: your service account. Without sharing, the client acts as the call enforces the boundary, so no bug in this library can hand someone an object they may not see. With sharing, your `Authorizer` is the only thing between a caller and the cache. -You can have both. Pair `SharedBackend` with [`kube.SSARAuthorizer`](../gateway/kube/authz.go), which +You can have both. Pair `SharedBackend` with [`kube.SubjectAccessReviewAuthorizer`](../gateway/kube/authz.go), which asks the API server through a SubjectAccessReview whether this user may list and watch this resource here, before serving them from the shared cache. Kubernetes decides again, per user, per snapshot cycle, at the cost of one round-trip. Read [auth.md](auth.md) before wiring it. diff --git a/examples/README.md b/examples/README.md index 13c05e6..5cab45b 100644 --- a/examples/README.md +++ b/examples/README.md @@ -9,3 +9,6 @@ For host integration patterns, use these small recipes: explicit `Authorization` header for a deliberate token-bearing client. - [Shared backend with SSAR](../docs/adopting.md#4-share-watches-only-with-kubernetes-backed-authorization): one service-account watch plus Kubernetes SubjectAccessReviews for each subscriber. + +[Conditional save](conditional-save/README.md) composes a managed connection with atomic save capture, +projected reconciliation, and a host-owned Kubernetes endpoint that preserves real 409 conflicts. diff --git a/examples/conditional-save/README.md b/examples/conditional-save/README.md new file mode 100644 index 0000000..34bec18 --- /dev/null +++ b/examples/conditional-save/README.md @@ -0,0 +1,43 @@ +# Conditional save with a live draft + +This example composes the existing store, managed stream and host-owned writes. It adds no merge +algorithm or shared watch implementation. + +- [editor.ts](editor.ts) captures a save intent synchronously, handles HTTP 409 by reconciling a + projected GET, retains in-flight edits and rejects late responses superseded by the watch. +- [handler.go](../../gateway/kube/examples/conditionalsave/handler.go) is a compilable ConfigMap + GET/PATCH endpoint. Mount it on a fixed namespace/name with a host-authenticated Kubernetes client + acting as the caller. The host owns credentials, CSRF protection, route authorization and audit. +- [saving tests](../../packages/krm-stream/test/saving.test.ts) exercise the response races. +- `TestConditionalSaveConflict` in the Kubernetes e2e suite exercises this handler against a real API + server: a competing update makes a captured save return 409 without overwriting the winner. + +```ts +const store = new LiveResourceStore(); +const connection = connectManagedResourceStream(streamURL, store, { + onStateChange: state => renderConnection(state), +}); +const editor = conditionalEditor(store, uid, "/editor/configmap", hostFetch); +const unsubscribe = store.subscribe(renderEditor); +// Enable Save only while connection.state.status === "live" and !editor.saving. +// On Save: await editor.save(), then render errors/conflicts and the current draft. +// On unmount: +unsubscribe(); +connection.close(); +``` + +Use the same scope and `krm-full/v1` projection on the stream and this example endpoint. The GET +returns a complete projected object with redactions; it must use a most-recent Kubernetes read, with +no HTTP or application cache. A recreated name has a different UID and must be opened as a new editor. +Keep an external draft archive if users need to recover edits after deletion: the store intentionally +removes drafts of deleted objects. + +A narrow merge patch is **not concurrency protection**. JSON merge patch replaces arrays as a whole, +even when the client merges keyed list items intelligently. This endpoint puts the captured UID and +resourceVersion into the Kubernetes patch. A stale version produces a safe 409, including when +invisible status changes were suppressed by the stream. Reconcile first; never transplant an old +patch onto the latest version. No automatic write retry is performed. + +Successful writes return 204. The stream echo settles the saved values while retaining later edits. +If the echo is delayed, dirty state remains visible; prevent repeated saves until your host's chosen +acknowledgment UX allows them. Save results never feed raw Kubernetes objects back into the store. diff --git a/examples/conditional-save/editor.ts b/examples/conditional-save/editor.ts new file mode 100644 index 0000000..11069bf --- /dev/null +++ b/examples/conditional-save/editor.ts @@ -0,0 +1,45 @@ +import type { KRMObject, LiveResourceStore, Redaction } from "../../packages/krm-stream/src/index.ts"; + +/** Copy into the host. request is the host's fetch wrapper (session/CSRF/error policy). + * Call save only while the managed connection is live. Keep the editor mounted during requests. */ +export function conditionalEditor(store: LiveResourceStore, uid: string, url: string, request: typeof fetch) { + let saving = false; + return { + get saving() { + return saving; + }, + async save(): Promise<"unchanged" | "busy" | "saved" | "conflict"> { + if (saving) return "busy"; + if (store.conflicts(uid).length) return "conflict"; + const intent = store.captureSave(uid); // No await between patch and merge-base capture. + if (!intent) return "unchanged"; + saving = true; + try { + const result = await request(url, { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(intent), + }); + if (result.status === 409) { + // Do not clear the draft or blindly retry the old patch using a newer version. + // A watch may already have deleted the object; don't resurrect it through a GET. + if (!store.ids().includes(uid)) return "conflict"; + const reconcile = store.captureReconciliation(uid); + const latest = await request(url, { cache: "no-store" }); + if (latest.status === 404) return "conflict"; // Let the stream remove the old UID. + if (!latest.ok) throw new Error(`Reconciliation failed: HTTP ${latest.status}`); + const body = (await latest.json()) as { object: KRMObject; redacted: Redaction[] | null }; + reconcile(body.object, { redacted: body.redacted ?? [] }); + // A newer watch/GET wins if reconcile returns false. Render the current store either way. + // Ask the user to review conflicts and save again; capture a fresh intent on that click. + return "conflict"; + } + if (!result.ok) throw new Error(`Save failed: HTTP ${result.status}`); + // No adoptSaved/clear/reset: even edits made while saving survive the watch echo. + return "saved"; + } finally { + saving = false; + } + }, + }; +} diff --git a/examples/vanilla-browser/tests/live-krm.spec.ts b/examples/vanilla-browser/tests/live-krm.spec.ts index 6d6c1ce..db77cd3 100644 --- a/examples/vanilla-browser/tests/live-krm.spec.ts +++ b/examples/vanilla-browser/tests/live-krm.spec.ts @@ -118,7 +118,10 @@ test("a redacted Secret value is not in the page at all — the mask is drawn, n await expect(page.getByTestId("patch")).toHaveText("null"); }); -test("a named object that does not exist renders as empty, not as a ghost and not as an error", async ({ page, visit }) => { +test("a named object that does not exist renders as empty, not as a ghost and not as an error", async ({ + page, + visit, +}) => { // reset, synced — and nothing else. The fixture that kills the "named scopes may skip the snapshot" // optimization: skip it, and a delete-while-disconnected leaves the object on screen forever. await visit("fixture=named-object-absent&pace=0ms"); @@ -126,3 +129,82 @@ test("a named object that does not exist renders as empty, not as a ghost and no await expect(page.getByTestId("editable-body")).toBeEmpty(); await expect(page.getByTestId("patch")).toHaveText("null"); }); + +test("managed fetch recovers a sequence gap and preserves the browser draft", async ({ page, entry }) => { + let attempts = 0; + const object = { + apiVersion: "v1", + kind: "ConfigMap", + metadata: { uid: "managed", name: "cm", resourceVersion: "1" }, + data: { value: "base" }, + }; + await page.route("**/managed-test", (route) => { + attempts++; + const events = + attempts === 1 + ? [ + { seq: 1, type: "reset" }, + { seq: 3, type: "synced" }, + ] + : [ + { seq: 1, type: "reset" }, + { seq: 2, type: "added", object }, + { seq: 3, type: "synced" }, + ]; + return route.fulfill({ + contentType: "text/event-stream", + body: events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), + }); + }); + await page.goto("/"); + const result = await page.evaluate( + async ({ entry, object }) => { + const lib = await import(`/krm-stream/${entry === "bundle" ? "krm-stream" : "index"}.js`); + const store = new lib.LiveResourceStore(); + store.applyServerEvent(object); + store.setValue("managed", ["data", "value"], "draft"); + const states: string[] = []; + const handle = lib.connectManagedResourceStream("/managed-test", store, { + maxRetries: 1, + retryDelayMs: 0, + onStateChange: (state: { status: string }) => states.push(state.status), + onSynced: () => handle.close(), + }); + await handle.closed; + return { states, draft: store.draft("managed").data.value, status: handle.state.status }; + }, + { entry, object }, + ); + expect(attempts).toBe(2); + expect(result.draft).toBe("draft"); + expect(result.states).toContain("retrying"); + expect(result.states).toContain("live"); + expect(result.status).toBe("closed"); +}); + +test("native EventSource resets its sequence on a browser reconnect", async ({ page, entry }) => { + let attempts = 0; + await page.route("**/native-reconnect-test", (route) => { + attempts++; + return route.fulfill({ + contentType: "text/event-stream", + body: 'retry: 1\n\ndata: {"seq":1,"type":"reset"}\n\ndata: {"seq":2,"type":"synced"}\n\n', + }); + }); + await page.goto("/"); + const result = await page.evaluate(async (entry) => { + const lib = await import(`/krm-stream/${entry === "bundle" ? "krm-stream" : "index"}.js`); + let synced = 0; + let gaps = 0; + const handle = lib.connectWithEventSource("/native-reconnect-test", new lib.LiveResourceStore(), { + onSynced: () => { + if (++synced === 2) handle.close(); + }, + onGap: () => gaps++, + }); + await handle.closed; + return { synced, gaps }; + }, entry); + expect(attempts).toBe(2); + expect(result).toEqual({ synced: 2, gaps: 0 }); +}); diff --git a/gateway/README.md b/gateway/README.md index b671bba..7a8fd2c 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -83,7 +83,7 @@ and projection decision, then call `ValidateMergePatch` before issuing a Kuberne ## Shared backends `SharedBackend` multiplexes one upstream watch per normalized scope. It is opt-in because the shared -watch uses one identity. Pair it with `kube.SSARAuthorizer` when Kubernetes should continue making +watch uses one identity. Pair it with `kube.SubjectAccessReviewAuthorizer` when Kubernetes should continue making per-caller access decisions. `SharedOptions.QueueDepth` bounds live events per slow subscriber. Overflow triggers a resnapshot; @@ -97,3 +97,9 @@ Observers run on stream paths, so they must return promptly. See [`docs/operations.md`](../docs/operations.md) for suggested metrics and alerts, and [`docs/adopting.md`](../docs/adopting.md) for full host wiring examples. + +For shared streams, set `Options.ReauthorizationInterval` (for example, 30 seconds) and +`Options.ReauthorizationTimeout` (for example, 5 seconds). Each subscriber is rechecked independently, +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`; `SSARAuthorizer` is a deprecated compatibility alias. diff --git a/gateway/handler.go b/gateway/handler.go index eb2f44d..8fb3b4b 100644 --- a/gateway/handler.go +++ b/gateway/handler.go @@ -68,6 +68,13 @@ type Options struct { // HeartbeatInterval defaults to HeartbeatInterval. Set a positive value to match a proxy's idle // timeout; it affects HTTP streams only. HeartbeatInterval time.Duration + + // ReauthorizationInterval rechecks each subscriber independently, even on quiet streams. + // Zero disables timed checks; snapshot cycles always reauthorize. + ReauthorizationInterval time.Duration + // ReauthorizationTimeout bounds a timed check. Zero defaults to 10 seconds. + // Authorizers and projection policies must honor context cancellation. + ReauthorizationTimeout time.Duration } // Handler mounts the stream on one route. @@ -89,13 +96,15 @@ func Handler(o Options) http.Handler { } g := &Gateway{ - Auth: o.Authorizer, - Clients: o.Clients, - Projection: o.Projection, - Projections: o.Projections, - Ordering: o.Ordering, - Observer: o.Observer, - HeartbeatInterval: o.HeartbeatInterval, + Auth: o.Authorizer, + Clients: o.Clients, + Projection: o.Projection, + Projections: o.Projections, + Ordering: o.Ordering, + Observer: o.Observer, + HeartbeatInterval: o.HeartbeatInterval, + ReauthorizationInterval: o.ReauthorizationInterval, + ReauthorizationTimeout: o.ReauthorizationTimeout, } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/gateway/kube/authz.go b/gateway/kube/authz.go index c34de30..801a22d 100644 --- a/gateway/kube/authz.go +++ b/gateway/kube/authz.go @@ -28,7 +28,7 @@ import ( // Kubernetes' again. That is the point: // // shared := gateway.NewSharedBackend(serviceAccountBackend) // one watch, one identity -// opts.Authorizer = kube.SSARAuthorizer(clientset, subjectOf) // …but RBAC still decides +// opts.Authorizer = kube.SubjectAccessReviewAuthorizer(clientset, subjectOf) // …but RBAC still decides // opts.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend, error) { return shared, nil } // // Because the gateway re-authorizes on every snapshot cycle (stream.go), this is also how a @@ -51,7 +51,7 @@ type Subject struct { // SubjectFor maps a Principal onto a Kubernetes subject. Return an error to refuse the caller. type SubjectFor func(gateway.Principal) (Subject, error) -// SSARAuthorizer authorizes a scope by asking the API server, with a SubjectAccessReview, whether the +// SubjectAccessReviewAuthorizer authorizes a scope by asking the API server, with a SubjectAccessReview, whether the // caller may `list` and `watch` that resource. // // BOTH verbs, and that is not belt-and-braces: a snapshot cycle is a list followed by a watch — quite @@ -62,7 +62,7 @@ type SubjectFor func(gateway.Principal) (Subject, error) // It needs the SERVER's own client (a service account) to hold `create` on `subjectaccessreviews`, // which is the standard `system:auth-delegator` role. It does NOT need impersonate rights: this asks // a question about a user, it does not act as one. -func SSARAuthorizer(cs kubernetes.Interface, subjectFor SubjectFor) gateway.Authorizer { +func SubjectAccessReviewAuthorizer(cs kubernetes.Interface, subjectFor SubjectFor) gateway.Authorizer { return gateway.AuthorizerFunc(func(ctx context.Context, p gateway.Principal, scope gateway.Scope) error { subject, err := subjectFor(p) if err != nil { @@ -131,3 +131,9 @@ func groupResource(s gateway.Scope) string { } return s.Group + "/" + s.Resource } + +// SSARAuthorizer is the historical name for SubjectAccessReviewAuthorizer. +// Deprecated: use SubjectAccessReviewAuthorizer; this creates SubjectAccessReview, not SelfSubjectAccessReview. +func SSARAuthorizer(cs kubernetes.Interface, subjectFor SubjectFor) gateway.Authorizer { + return SubjectAccessReviewAuthorizer(cs, subjectFor) +} diff --git a/gateway/kube/e2e_test.go b/gateway/kube/e2e_test.go index 7955992..1301086 100644 --- a/gateway/kube/e2e_test.go +++ b/gateway/kube/e2e_test.go @@ -12,8 +12,12 @@ package kube_test import ( + "bytes" "context" + "encoding/json" "errors" + "net/http" + "net/http/httptest" "testing" "time" @@ -29,6 +33,7 @@ import ( "github.com/ConfigButler/krm-stream/gateway" "github.com/ConfigButler/krm-stream/gateway/kube" + "github.com/ConfigButler/krm-stream/gateway/kube/examples/conditionalsave" ) var flunders = schema.GroupVersionResource{Group: "wardle.example.com", Version: "v1alpha1", Resource: "flunders"} @@ -259,3 +264,41 @@ func TestRealClusterAggregatedAPIFallsBack(t *testing.T) { t.Fatalf("deleted carried no trustworthy uid: %+v", deleted.Identity) } } + +// This uses the real API server's optimistic concurrency implementation, not fake client reactors. +func TestConditionalSaveConflict(t *testing.T) { + cs, _ := clients(t) + ns := scratchNamespace(t, cs) + ctx := t.Context() + cm, err := cs.CoreV1().ConfigMaps(ns).Create(ctx, &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "editor"}, Data: map[string]string{"value": "base"}}, metav1.CreateOptions{}) + if err != nil { + t.Fatal(err) + } + handler := conditionalsave.Handler(func(*http.Request) (kubernetes.Interface, error) { return cs, nil }, ns, cm.Name) + captured := map[string]any{"uid": string(cm.UID), "resourceVersion": cm.ResourceVersion, "patch": map[string]any{"data": map[string]any{"value": "mine"}}} + cm.Data["value"] = "winner" + latest, err := cs.CoreV1().ConfigMaps(ns).Update(ctx, cm, metav1.UpdateOptions{}) + if err != nil { + t.Fatal(err) + } + send := func() int { + body, err := json.Marshal(captured) + if err != nil { + t.Fatal(err) + } + response := httptest.NewRecorder() + handler.ServeHTTP(response, httptest.NewRequest(http.MethodPatch, "/editor", bytes.NewReader(body))) + return response.Code + } + if code := send(); code != http.StatusConflict { + t.Fatalf("stale save: HTTP %d, want real Kubernetes 409", code) + } + got, err := cs.CoreV1().ConfigMaps(ns).Get(ctx, cm.Name, metav1.GetOptions{}) + if err != nil || got.Data["value"] != "winner" { + t.Fatalf("winner overwritten: %v, %v", got, err) + } + captured["resourceVersion"] = latest.ResourceVersion // Represents a newly reviewed/captured intent. + if code := send(); code != http.StatusNoContent { + t.Fatalf("fresh save: HTTP %d", code) + } +} diff --git a/gateway/kube/examples/conditionalsave/handler.go b/gateway/kube/examples/conditionalsave/handler.go new file mode 100644 index 0000000..22c56d2 --- /dev/null +++ b/gateway/kube/examples/conditionalsave/handler.go @@ -0,0 +1,137 @@ +// Package conditionalsave is a host-owned ConfigMap editor endpoint example. Copy and adapt it; +// this is intentionally outside the library API. Mount it on a fixed, host-authorized namespace/name. +package conditionalsave + +import ( + "encoding/json" + "errors" + "io" + "net/http" + + "github.com/ConfigButler/krm-stream/gateway" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" +) + +type saveRequest struct { + UID string `json:"uid"` + ResourceVersion string `json:"resourceVersion"` + Patch map[string]any `json:"patch"` +} + +// Handler uses a host-authenticated, caller-scoped Kubernetes client on every request. The host +// owns session validation, CSRF protection, routing and audit. This example pins ProjectionFull, +// which must also be the stream's effective projection. clientFor must enforce access to this route. +func Handler(clientFor func(*http.Request) (kubernetes.Interface, error), namespace, name string) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + if r.Method != http.MethodGet && r.Method != http.MethodPatch { + w.Header().Set("Allow", "GET, PATCH") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + cs, err := clientFor(r) + if err != nil { + http.Error(w, "not authenticated", http.StatusUnauthorized) + return + } + client := cs.CoreV1().ConfigMaps(namespace) + current, err := client.Get(r.Context(), name, metav1.GetOptions{}) + if err != nil { + writeError(w, err) + return + } + object, err := runtime.DefaultUnstructuredConverter.ToUnstructured(current) + if err != nil { + http.Error(w, "conversion failed", http.StatusInternalServerError) + return + } + if r.Method == http.MethodGet { + projected, _ := gateway.Project(gateway.ProjectionFull, object) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(struct { + Object gateway.KRMObject `json:"object"` + Redacted []gateway.Redaction `json:"redacted"` + }{projected, []gateway.Redaction{}}) + return + } + var request saveRequest + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)) + decoder.DisallowUnknownFields() + if err = decoder.Decode(&request); err != nil { + http.Error(w, "invalid save request", http.StatusBadRequest) + return + } + if err = decoder.Decode(new(any)); err != io.EOF { + http.Error(w, "expected one save request", http.StatusBadRequest) + return + } + if request.UID == "" || request.ResourceVersion == "" || request.Patch == nil { + http.Error(w, "UID, resourceVersion and patch are required", http.StatusBadRequest) + return + } + if request.UID != string(current.UID) { + http.Error(w, "object was replaced", http.StatusConflict) + return + } + // Application edit policy is separate from projection protection. Metadata identity and version + // are never accepted inside the user's patch; the host sets the captured preconditions below. + for key, value := range request.Patch { + switch key { + case "data", "binaryData": + case "metadata": + metadata, ok := value.(map[string]any) + if !ok { + http.Error(w, "metadata must be an object", http.StatusBadRequest) + return + } + for field := range metadata { + if field != "labels" && field != "annotations" { + http.Error(w, "metadata field is not editable", http.StatusBadRequest) + return + } + } + default: + http.Error(w, "field is not editable", http.StatusBadRequest) + return + } + } + metadata, _ := request.Patch["metadata"].(map[string]any) + if metadata == nil { + metadata = map[string]any{} + request.Patch["metadata"] = metadata + } + metadata["uid"] = request.UID + metadata["resourceVersion"] = request.ResourceVersion // NEVER current.ResourceVersion + patch, err := json.Marshal(request.Patch) + if err != nil { + http.Error(w, "invalid patch", http.StatusBadRequest) + return + } + if err = gateway.ValidateMergePatch(gateway.ProjectionFull, object, patch); err != nil { + http.Error(w, "patch touches a protected field", http.StatusBadRequest) + return + } + _, err = client.Patch(r.Context(), name, types.MergePatchType, patch, metav1.PatchOptions{}) + if err != nil { + writeError(w, err) + return + } + w.WriteHeader(http.StatusNoContent) // The projected watch echo settles dirty state. + }) +} + +func writeError(w http.ResponseWriter, err error) { + status := http.StatusBadGateway + var apiStatus apierrors.APIStatus + if errors.As(err, &apiStatus) { + status = int(apiStatus.Status().Code) + } + if status < 400 || status > 599 { + status = http.StatusBadGateway + } + http.Error(w, http.StatusText(status), status) +} diff --git a/gateway/kube/examples/conditionalsave/handler_test.go b/gateway/kube/examples/conditionalsave/handler_test.go new file mode 100644 index 0000000..771faa8 --- /dev/null +++ b/gateway/kube/examples/conditionalsave/handler_test.go @@ -0,0 +1,62 @@ +package conditionalsave + +import ( + "bytes" + "net/http" + "net/http/httptest" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +func TestHostPreservesCapturedVersionAndKubernetesConflict(t *testing.T) { + cs := fake.NewSimpleClientset(&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "cm", Namespace: "app", UID: "u", ResourceVersion: "2"}}) + var sent []byte + cs.PrependReactor("patch", "configmaps", func(action k8stesting.Action) (bool, runtime.Object, error) { + sent = action.(k8stesting.PatchAction).GetPatch() + return true, nil, apierrors.NewConflict(schema.GroupResource{Resource: "configmaps"}, "cm", nil) + }) + handler := Handler(func(*http.Request) (kubernetes.Interface, error) { return cs, nil }, "app", "cm") + response := httptest.NewRecorder() + handler.ServeHTTP(response, httptest.NewRequest(http.MethodPatch, "/", strings.NewReader(`{"uid":"u","resourceVersion":"1","patch":{"data":{"value":"mine"}}}`))) + if response.Code != http.StatusConflict { + t.Fatalf("got HTTP %d", response.Code) + } + if !bytes.Contains(sent, []byte(`"resourceVersion":"1"`)) || !bytes.Contains(sent, []byte(`"uid":"u"`)) { + t.Fatalf("captured preconditions lost: %s", sent) + } +} + +func TestHostRejectsIdentityAndProjectionBypasses(t *testing.T) { + for _, body := range []string{ + `{"uid":"u","resourceVersion":"1","patch":{"metadata":null}}`, + `{"uid":"u","resourceVersion":"1","patch":{"metadata":{"resourceVersion":"2"}}}`, + `{"uid":"u","resourceVersion":"1","patch":{"metadata":{"annotations":{"kubectl.kubernetes.io/last-applied-configuration":"secret"}}}}`, + `{"uid":"u","resourceVersion":"1","patch":{"status":{}}}`, + `{"uid":"u","patch":{"data":{"value":"mine"}}}`, + `{"uid":"u","resourceVersion":"1","patch":{}} {}`, + } { + t.Run(body, func(t *testing.T) { + cs := fake.NewSimpleClientset(&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "cm", Namespace: "app", UID: "u", ResourceVersion: "1"}}) + handler := Handler(func(*http.Request) (kubernetes.Interface, error) { return cs, nil }, "app", "cm") + response := httptest.NewRecorder() + handler.ServeHTTP(response, httptest.NewRequest(http.MethodPatch, "/", strings.NewReader(body))) + if response.Code != http.StatusBadRequest { + t.Fatalf("got HTTP %d", response.Code) + } + for _, action := range cs.Actions() { + if action.GetVerb() == "patch" { + t.Fatal("rejected body reached Kubernetes PATCH") + } + } + }) + } +} diff --git a/gateway/reauthorize.go b/gateway/reauthorize.go new file mode 100644 index 0000000..7c3f4b8 --- /dev/null +++ b/gateway/reauthorize.go @@ -0,0 +1,79 @@ +package gateway + +import ( + "context" + "sync" + "time" +) + +// The gate serializes disclosure with reauthorization. While a check is pending, this subscriber +// receives no objects. Other subscribers and the shared upstream continue independently. +type authorizationSink struct { + mu sync.Mutex + sink Sink +} + +func (s *authorizationSink) Emit(ctx context.Context, event Event) error { + s.mu.Lock() + defer s.mu.Unlock() + if err := ctx.Err(); err != nil { + return err + } + return s.sink.Emit(ctx, event) +} + +func (g *Gateway) authorizedCycle(ctx context.Context, principal Principal, scope Scope, requested Projection, policy ProjectionPolicy, backend Backend, projection Projection, revisions map[string]map[string]redactionState, sink Sink) error { + if g.ReauthorizationInterval <= 0 { + return g.cycle(ctx, backend, scope, projection, revisions, sink) + } + cycleCtx, cancel := context.WithCancel(ctx) + defer cancel() + gate := &authorizationSink{sink: sink} + done := make(chan error, 1) + go func() { + ticker := time.NewTicker(g.ReauthorizationInterval) + defer ticker.Stop() + for { + select { + case <-cycleCtx.Done(): + done <- nil + return + case <-ticker.C: + gate.mu.Lock() + timeout := g.ReauthorizationTimeout + if timeout <= 0 { + timeout = 10 * time.Second + } + checkCtx, finish := context.WithTimeout(cycleCtx, timeout) + err := g.Auth.Authorize(checkCtx, principal, scope) + if err == nil { + var selected Projection + selected, err = policy.SelectProjection(checkCtx, principal, scope, requested) + if err == nil && selected != projection { + err = Forbidden("stream projection authorization changed; open a new stream") + } + } + if err == nil { + err = checkCtx.Err() + } + finish() + if err != nil { + // Fail closed even if a host returns a normally recoverable StreamError. + se := *asStreamError(err) + se.Terminal = true + cancel() + gate.mu.Unlock() + done <- &se + return + } + gate.mu.Unlock() + } + } + }() + err := g.cycle(cycleCtx, backend, scope, projection, revisions, gate) + cancel() + if authErr := <-done; authErr != nil { + return authErr + } + return err +} diff --git a/gateway/reauthorize_test.go b/gateway/reauthorize_test.go new file mode 100644 index 0000000..0cdd9e5 --- /dev/null +++ b/gateway/reauthorize_test.go @@ -0,0 +1,167 @@ +package gateway + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +type reauthEvents chan Event + +func (s reauthEvents) Emit(ctx context.Context, ev Event) error { + select { + case s <- ev: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} +func awaitEvent(t *testing.T, events reauthEvents, kind EventType) Event { + t.Helper() + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + for { + select { + case ev := <-events: + if ev.Type == kind { + return ev + } + case <-timer.C: + t.Fatalf("waiting for %s", kind) + } + } +} + +func TestTimedRevocationOnlyDisconnectsDeniedSharedSubscriber(t *testing.T) { + upstream := newFakeUpstream() + shared := NewSharedBackend(upstream) + var revoked atomic.Bool + g := &Gateway{ + Auth: AuthorizerFunc(func(_ context.Context, p Principal, _ Scope) error { + if p == "alice" && revoked.Load() { + return Forbidden("revoked") + } + return nil + }), + Clients: func(context.Context, string, Principal) (Backend, error) { return shared, nil }, + ReauthorizationInterval: 5 * time.Millisecond, + } + alice, bob := make(reauthEvents, 32), make(reauthEvents, 32) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + aDone, bDone := make(chan error, 1), make(chan error, 1) + go func() { aDone <- g.Stream(ctx, "alice", sharedScopeUnderTest, alice) }() + go func() { bDone <- g.Stream(ctx, "bob", sharedScopeUnderTest, bob) }() + upstream.send(WatchEvent{Type: WatchBookmark, InitialEventsEnd: true}) + awaitEvent(t, alice, EventSynced) + awaitEvent(t, bob, EventSynced) + revoked.Store(true) + denied := awaitEvent(t, alice, EventError) + if denied.Code != CodeForbidden || !denied.Terminal { + t.Fatalf("not terminal denial: %+v", denied) + } + <-aDone + upstream.send(WatchEvent{Type: WatchAdded, Object: obj("u", "cm", "1")}) + awaitEvent(t, bob, EventAdded) + if upstream.opens() != 1 { + t.Fatal("revocation reopened shared upstream") + } + cancel() + <-bDone +} + +func TestTimedAuthorizationTimeoutFailsClosed(t *testing.T) { + var calls atomic.Int32 + upstream := newFakeUpstream() + g := &Gateway{ + Auth: AuthorizerFunc(func(ctx context.Context, _ Principal, _ Scope) error { + if calls.Add(1) == 1 { + return nil + } + <-ctx.Done() + return ctx.Err() + }), + Clients: func(context.Context, string, Principal) (Backend, error) { return upstream, nil }, + ReauthorizationInterval: time.Millisecond, ReauthorizationTimeout: 5 * time.Millisecond, + } + events := make(reauthEvents, 32) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- g.Stream(ctx, "alice", sharedScopeUnderTest, events) }() + ev := awaitEvent(t, events, EventError) + if !ev.Terminal || ev.Code != CodeInternal { + t.Fatalf("timeout did not fail closed: %+v", ev) + } + <-done +} + +func TestTimedProjectionWithdrawalTerminatesStream(t *testing.T) { + var checks atomic.Int32 + g := &Gateway{ + Auth: AllowAll{}, + Projections: ProjectionPolicyFunc(func(context.Context, Principal, Scope, Projection) (Projection, error) { + if checks.Add(1) == 1 { + return ProjectionRaw, nil + } + return ProjectionFull, nil + }), + Clients: func(context.Context, string, Principal) (Backend, error) { return newFakeUpstream(), nil }, + ReauthorizationInterval: time.Millisecond, + } + events := make(reauthEvents, 32) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- g.Stream(ctx, "alice", sharedScopeUnderTest, events) }() + ev := awaitEvent(t, events, EventError) + if ev.Code != CodeForbidden || !ev.Terminal { + t.Fatalf("projection change kept old view: %+v", ev) + } + <-done +} + +func TestTimedCheckPausesDisclosureAndCancellationStopsCheck(t *testing.T) { + upstream := newFakeUpstream() + var calls atomic.Int32 + checking := make(chan struct{}) + canceled := make(chan struct{}) + g := &Gateway{ + Auth: AuthorizerFunc(func(ctx context.Context, _ Principal, _ Scope) error { + if calls.Add(1) == 1 { + return nil + } + close(checking) + <-ctx.Done() + close(canceled) + return ctx.Err() + }), + Clients: func(context.Context, string, Principal) (Backend, error) { return upstream, nil }, + ReauthorizationInterval: time.Millisecond, + } + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + events := make(reauthEvents, 32) + done := make(chan error, 1) + go func() { done <- g.Stream(ctx, "alice", sharedScopeUnderTest, events) }() + awaitEvent(t, events, EventReset) + select { + case <-checking: + case <-ctx.Done(): + t.Fatal("check never started") + } + upstream.send(WatchEvent{Type: WatchAdded, Object: obj("u", "cm", "1")}) + select { + case ev := <-events: + t.Fatalf("disclosed while authorization pending: %+v", ev) + case <-time.After(10 * time.Millisecond): + } + cancel() + <-done + select { + case <-canceled: + default: + t.Fatal("check outlived stream") + } +} diff --git a/gateway/shared.go b/gateway/shared.go index c8efa3f..fc71216 100644 --- a/gateway/shared.go +++ b/gateway/shared.go @@ -30,10 +30,10 @@ import ( // and there is a way to opt in WITHOUT giving up the boundary, which is the wiring you want: // // shared := gateway.NewSharedBackend(myServiceAccountBackend) // one watch, ONE identity… -// opts.Authorizer = kube.SSARAuthorizer(clientset, subjectOf) // …but Kubernetes still decides +// opts.Authorizer = kube.SubjectAccessReviewAuthorizer(clientset, subjectOf) // …but Kubernetes still decides // opts.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend, error) { return shared, nil } // -// kube.SSARAuthorizer asks the API server, with a SubjectAccessReview, whether THIS user may list and +// kube.SubjectAccessReviewAuthorizer asks the API server, with a SubjectAccessReview, whether THIS user may list and // watch THIS resource here — before the subscriber is served from the shared cache. RBAC is the // boundary again, and the sharing costs you nothing but a round-trip per snapshot cycle. See // docs/auth.md. diff --git a/gateway/stream.go b/gateway/stream.go index a6b09cb..627b6c5 100644 --- a/gateway/stream.go +++ b/gateway/stream.go @@ -90,6 +90,13 @@ 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 + + // ReauthorizationInterval rechecks each subscriber independently, even on quiet streams. + // Zero disables timed checks; snapshot cycles always reauthorize. + ReauthorizationInterval time.Duration + // ReauthorizationTimeout bounds a timed check. Zero defaults to 10 seconds. + // Authorizers and projection policies must honor context cancellation. + ReauthorizationTimeout time.Duration // Ordering is how far the upstream's resourceVersions may be trusted. The zero value is // OrderingStrict: this library targets Kubernetes 1.35+, where orderability is a conformance // requirement, and it says so rather than degrading quietly on every cluster to accommodate one. @@ -141,11 +148,6 @@ func (g *Gateway) StreamProjection(ctx context.Context, principal Principal, sco // EventSource reconnects on its own, so a non-terminal refusal would have a revoked user // hammering a forbidden scope forever. // - // What this does NOT do, and docs/auth.md says so plainly: a perfectly quiet stream may not - // cycle for a long time, so revocation is noticed at the next cycle rather than instantly. - // The credential problem is solved properly on the other side of the seam — ClientFor is - // re-invoked here too, so a host that returns a client backed by a refreshing token source - // (the Dex route) never hands us a dead token in the first place. if err := g.Auth.Authorize(ctx, principal, scope); err != nil { return g.emitTerminal(ctx, sink, scope, "", err) } @@ -162,7 +164,7 @@ func (g *Gateway) StreamProjection(ctx context.Context, principal Principal, sco return g.emitTerminal(ctx, sink, scope, projection, err) } - err = g.cycle(ctx, backend, scope, projection, revisions, sink) + err = g.authorizedCycle(ctx, principal, scope, requested, policy, backend, projection, revisions, sink) switch { case err == nil: // A cycle only ends by error or cancellation; nil would be a bug in the loop below. diff --git a/packages/krm-stream/README.md b/packages/krm-stream/README.md index 04ab390..597f1ea 100644 --- a/packages/krm-stream/README.md +++ b/packages/krm-stream/README.md @@ -4,7 +4,8 @@ stream in a browser or JavaScript application. It provides: - `LiveResourceStore` for server state, local drafts, conflicts, redactions, and merge patches. -- `connectWithEventSource` for same-origin browser streams. +- `connectManagedResourceStream` for bounded recovery with cookies or bearer headers. +- `connectWithEventSource` for low-level native EventSource integration. - `connectResourceStream` for fetch-based transports with explicit headers. - `resourceStreamURL` for the v1 scope query format. @@ -12,10 +13,10 @@ The package is headless and does not choose a UI framework. It works with the Go repository or any conforming producer. ```ts -import { LiveResourceStore, connectWithEventSource, resourceStreamURL } from "@configbutler/krm-stream"; +import { LiveResourceStore, connectManagedResourceStream, resourceStreamURL } from "@configbutler/krm-stream"; const store = new LiveResourceStore(); -connectWithEventSource( +connectManagedResourceStream( resourceStreamURL("/resource-stream/v1", { version: "v1", resource: "configmaps", namespace: "app" }), store, ); @@ -37,7 +38,7 @@ import { LiveResourceStore } from "@configbutler/krm-stream/bundle"; Vendored, that is one file to copy (`dist/krm-stream.js`) and one path to serve: ```js -import { LiveResourceStore, connectWithEventSource } from "/krm-stream/krm-stream.js"; +import { LiveResourceStore, connectManagedResourceStream } from "/krm-stream/krm-stream.js"; ``` Both entry points are built from the same source and are exercised by the same browser test suite @@ -48,3 +49,40 @@ against a real `EventSource`. Neither has a runtime dependency. The unscoped `krm-stream` package is a compatibility forwarder. This project is pre-1.0: the protocol and the API may still change before 1.0. See the repository [README](../../README.md), [client state model](../../docs/client-state-model.md), and [release guide](../../docs/releasing.md). + +## Managed connections and conditional saves + +```ts +const connection = connectManagedResourceStream(url, store, { + maxRetries: 8, + onStateChange: state => renderConnection(state.status), + // headers: { Authorization: `Bearer ${hostToken}` }, // optional; credentials stay host-owned +}); +const unsubscribe = connection.subscribe(state => renderConnection(state.status)); +// On teardown: +unsubscribe(); +connection.close(); +await connection.closed; +``` + +Import `connectManagedResourceStream` from the package. It uses fetch for same-origin cookies or +bearer headers; `credentials: "include"` opts into cross-origin cookies. It requests a fresh snapshot +on sequence gaps, network failures, HTTP 408/429/5xx and EOF. Existing drafts survive recovery. +States are `connecting`, `syncing`, `live`, `retrying`, `closed`, `terminal`, and `exhausted`. +A reset makes the connection `syncing` until `synced`; enable saves while `live`. + +Defaults: eight retries **total per handle**, exponential delay from 500ms capped at 30s, with jitter. +Short successful connections do not replenish the budget. Terminal protocol errors and HTTP client +errors (including 401/403, excluding 408/429) stop retries. `close()` or `signal` cancels the stream and +pending backoff; `closed` resolves after cleanup. Create a new handle after credentials change or an +explicit user retry. The low-level fetch and native EventSource connectors remain available; native +EventSource owns network reconnects but closes on sequence gaps. Use the managed connector for +bounded recovery and observable lifecycle state. + +`store.captureSave(uid)` captures a detached patch, UID and base resourceVersion together. +`store.captureReconciliation(uid)` guards a projected asynchronous response against newer watch state. +See the [complete conditional-save example](../../examples/conditional-save/README.md) and +[save guide](../../docs/saving.md) for host preconditions, real Kubernetes 409s, and draft preservation. + +For Vue 3, use the [copyable composable](../../docs/vue.md) for reactive resource and connection state +with automatic subscription cleanup. Vue stays in the host application. diff --git a/packages/krm-stream/src/connection.ts b/packages/krm-stream/src/connection.ts new file mode 100644 index 0000000..4b244b6 --- /dev/null +++ b/packages/krm-stream/src/connection.ts @@ -0,0 +1,132 @@ +import { connectResourceStream, type StreamHandle, type StreamOptions } from "./sse.ts"; +import type { LiveResourceStore } from "./store.ts"; + +export type ConnectionStatus = "connecting" | "syncing" | "live" | "retrying" | "closed" | "terminal" | "exhausted"; + +export interface ConnectionState { + status: ConnectionStatus; + /** Retries used during this connection's lifetime (never reset by a short-lived success). */ + retries: number; + retryInMs?: number; +} + +export interface ManagedStreamOptions extends StreamOptions { + /** Total retry budget. Defaults to 8. Start a new connection to retry after exhaustion. */ + maxRetries?: number; + /** Exponential backoff starts at 500ms, capped at 30s, with 50–100% jitter. */ + retryDelayMs?: number; + maxRetryDelayMs?: number; + onStateChange?: (state: Readonly) => void; +} + +export interface ManagedStreamHandle extends StreamHandle { + readonly state: Readonly; + subscribe(callback: (state: Readonly) => void): () => void; +} + +/** Managed fetch transport for both session cookies and bearer headers. Every reconnect requests a + * fresh snapshot; the existing store and its drafts survive. HTTP 401/403 and terminal protocol + * errors stop permanently. EOF, network failures and sequence gaps consume a bounded retry budget. */ +export function connectManagedResourceStream( + url: string, + store: LiveResourceStore, + opts: ManagedStreamOptions = {}, +): ManagedStreamHandle { + const maxRetries = opts.maxRetries ?? 8; + const delay = opts.retryDelayMs ?? 500; + const cap = opts.maxRetryDelayMs ?? 30_000; + if ( + !Number.isSafeInteger(maxRetries) || + maxRetries < 0 || + !Number.isFinite(delay) || + delay < 0 || + !Number.isFinite(cap) || + cap < 0 || + cap > 2_147_483_647 + ) { + throw new RangeError("krm-stream: invalid retry budget or delay"); + } + const controller = new AbortController(); + const subscribers = new Set<(state: Readonly) => void>(); + let state: Readonly = Object.freeze({ status: "connecting", retries: 0 }); + let terminal = false; + const publish = (status: ConnectionStatus, retryInMs?: number) => { + state = Object.freeze({ status, retries: state.retries, ...(retryInMs === undefined ? {} : { retryInMs }) }); + opts.onStateChange?.(state); + for (const callback of subscribers) callback(state); + }; + const close = () => controller.abort(); + opts.signal?.addEventListener("abort", close, { once: true }); + if (opts.signal?.aborted) close(); + + // Defer opening until the handle exists, so subscribers can observe the first transition. + const closed = Promise.resolve().then(async () => { + try { + while (!controller.signal.aborted) { + publish("connecting"); + if (controller.signal.aborted) break; + const stream = connectResourceStream(url, store, { + ...opts, + signal: controller.signal, + onOpen: () => { + publish("syncing"); + opts.onOpen?.(); + }, + onChange: (change) => { + if (change.type === "reset") publish("syncing"); + opts.onChange?.(change); + }, + onSynced: () => { + publish("live"); + opts.onSynced?.(); + }, + onError: (code, message, isTerminal) => { + terminal ||= isTerminal; + opts.onError?.(code, message, isTerminal); + }, + }); + await stream.closed; + if (controller.signal.aborted) break; + if (terminal) { + publish("terminal"); + return; + } + if (state.retries >= maxRetries) { + publish("exhausted"); + return; + } + const ceiling = Math.min(cap, delay * 2 ** Math.min(state.retries, 30)); + const wait = Math.floor(ceiling * (0.5 + Math.random() * 0.5)); + state = { ...state, retries: state.retries + 1 }; + publish("retrying", wait); + await new Promise((resolve) => { + const done = () => { + clearTimeout(timer); + controller.signal.removeEventListener("abort", done); + resolve(); + }; + const timer = setTimeout(done, wait); + controller.signal.addEventListener("abort", done, { once: true }); + if (controller.signal.aborted) done(); + }); + } + publish("closed"); + } finally { + opts.signal?.removeEventListener("abort", close); + subscribers.clear(); + } + }); + return { + close, + closed, + get state() { + return state; + }, + subscribe(callback) { + subscribers.add(callback); + return () => { + subscribers.delete(callback); + }; + }, + }; +} diff --git a/packages/krm-stream/src/index.ts b/packages/krm-stream/src/index.ts index 7bb1467..1e57f61 100644 --- a/packages/krm-stream/src/index.ts +++ b/packages/krm-stream/src/index.ts @@ -14,16 +14,17 @@ // No runtime dependencies, and none of this knows anything about GitOps, Flux, Dex, kcp or // ConfigButler. It knows KRM. +export type { ConnectionState, ConnectionStatus, ManagedStreamHandle, ManagedStreamOptions } from "./connection.ts"; +export { connectManagedResourceStream } from "./connection.ts"; // Useful to a host that renders paths, and to anyone writing a policy: identity is a segment ARRAY. export { clone, deepEqual } from "./deep.ts"; export { get, has, isPrefix, parsePointer, pathKey } from "./path.ts"; - export { DEFAULT_EDITABLE_REGIONS, defaultPolicy, readOnlyPolicy, regionPolicy } from "./policy.ts"; export type { KubernetesStructuralSchema } from "./schema.ts"; export { withOpenAPIKeyedLists } from "./schema.ts"; export type { StreamChange, StreamHandle, StreamOptions } from "./sse.ts"; export { applyStreamEvent, connectResourceStream, connectWithEventSource, SSEDecoder, StreamSequence } from "./sse.ts"; -export type { ApplyOptions, ApplyResult } from "./store.ts"; +export type { ApplyOptions, ApplyResult, SaveRequest } from "./store.ts"; export { LiveResourceStore } from "./store.ts"; export type { Change, diff --git a/packages/krm-stream/src/sse.ts b/packages/krm-stream/src/sse.ts index dad1872..9fb887f 100644 --- a/packages/krm-stream/src/sse.ts +++ b/packages/krm-stream/src/sse.ts @@ -145,8 +145,11 @@ export interface StreamChange { } export interface StreamOptions { - /** Called for every `error` event. A terminal one has already closed the connection by the time - * this returns — there is nothing to retry, and retrying is the bug. */ + /** Transport connected; the snapshot may still be incomplete. */ + onOpen?: () => void; + /** Defaults to same-origin; use include for a cross-origin cookie gateway. */ + credentials?: RequestCredentials; + /** Called for protocol and HTTP errors. A terminal error ends the connection without retry. */ onError?: (code: ErrorCode, message: string, terminal: boolean) => void; /** Called at the end of every snapshot cycle. The store is now consistent: a good moment to paint. */ onSynced?: () => void; @@ -173,7 +176,8 @@ export function connectResourceStream(url: string, store: LiveResourceStore, opt if (opts.signal?.aborted) return { close: () => {}, closed: Promise.resolve() }; const controller = new AbortController(); const fetchImpl = opts.fetch ?? globalThis.fetch; - opts.signal?.addEventListener("abort", () => controller.abort(), { once: true }); + const abort = () => controller.abort(); + opts.signal?.addEventListener("abort", abort, { once: true }); const closed = (async () => { const res = await fetchImpl(url, { @@ -181,20 +185,35 @@ export function connectResourceStream(url: string, store: LiveResourceStore, opt headers: { Accept: "text/event-stream", ...opts.headers }, // The stream IS the response body; a cached one is a stream that never moves. cache: "no-store", + credentials: opts.credentials ?? "same-origin", }); if (!res.ok || !res.body) { - opts.onError?.("INTERNAL", `stream: HTTP ${res.status}`, true); + const code = res.status === 401 ? "UNAUTHENTICATED" : res.status === 403 ? "FORBIDDEN" : "INTERNAL"; + const terminal = res.status >= 400 && res.status < 500 && res.status !== 408 && res.status !== 429; + await res.body?.cancel(); + opts.onError?.(code, `stream: HTTP ${res.status}`, terminal); return; } + if (controller.signal.aborted) { + await res.body.cancel(); + return; + } + opts.onOpen?.(); const reader = res.body.pipeThrough(new TextDecoderStream()).getReader(); + const cancelReader = () => { + void reader.cancel().catch(() => {}); + }; + controller.signal.addEventListener("abort", cancelReader, { once: true }); + if (controller.signal.aborted) cancelReader(); const decoder = new SSEDecoder(); const sequence = new StreamSequence(); try { for (;;) { const { done, value } = await reader.read(); - if (done) return; + if (done || controller.signal.aborted) return; for (const ev of decoder.push(value)) { + if (controller.signal.aborted) return; if (feed(store, sequence, ev, opts)) { controller.abort(); // terminal: stop, and do NOT come back return; @@ -204,13 +223,14 @@ export function connectResourceStream(url: string, store: LiveResourceStore, opt } catch (err) { if (!controller.signal.aborted) throw err; } finally { - reader.cancel().catch(() => {}); + controller.signal.removeEventListener("abort", cancelReader); + await reader.cancel().catch(() => {}); } })(); return { close: () => controller.abort(), - closed: closed.catch(() => {}), + closed: closed.catch(() => {}).finally(() => opts.signal?.removeEventListener("abort", abort)), }; } @@ -220,22 +240,32 @@ export function connectResourceStream(url: string, store: LiveResourceStore, opt export function connectWithEventSource( url: string, store: LiveResourceStore, - opts: Omit = {}, + opts: Omit = {}, ): StreamHandle { if (opts.signal?.aborted) return { close: () => {}, closed: Promise.resolve() }; const es = new EventSource(url, { withCredentials: true }); - const sequence = new StreamSequence(); + let sequence = new StreamSequence(); + let stopped = false; let resolve: () => void; const closed = new Promise((r) => { resolve = r; }); const shut = () => { + stopped = true; es.close(); + opts.signal?.removeEventListener("abort", shut); resolve(); }; + es.onopen = () => { + if (stopped) return; + sequence = new StreamSequence(); + opts.onOpen?.(); + }; + es.onmessage = (e: MessageEvent) => { + if (stopped) return; let ev: StreamEvent; try { ev = JSON.parse(e.data) as StreamEvent; @@ -246,10 +276,10 @@ export function connectWithEventSource( }; // EventSource's `error` is also fired on a transport hiccup, where its OWN reconnect is the - // correct behaviour and we must not interfere. Only a terminal PROTOCOL error (which arrives as a - // message, above) closes the connection — and it must, or this reconnects forever. + // correct behaviour and we must not interfere. Terminal protocol errors and sequence gaps close + // this low-level handle. Use connectManagedResourceStream for managed gap recovery. es.onerror = () => { - if (es.readyState === EventSource.CLOSED) resolve(); + if (es.readyState === EventSource.CLOSED) shut(); }; opts.signal?.addEventListener("abort", shut, { once: true }); diff --git a/packages/krm-stream/src/store.ts b/packages/krm-stream/src/store.ts index b500691..2fda259 100644 --- a/packages/krm-stream/src/store.ts +++ b/packages/krm-stream/src/store.ts @@ -37,7 +37,15 @@ export interface ApplyOptions { redacted?: (Redaction | { path: Path; rev: number })[]; } +/** Detached save intent. The host adds these identity/version preconditions to the Kubernetes patch. */ +export interface SaveRequest { + uid: string; + resourceVersion: string; + patch: Record; +} + interface Resource { + revision: number; server: KRMObject; draft: KRMObject; redacted: { path: Path; rev: number }[]; @@ -46,6 +54,8 @@ interface Resource { export class LiveResourceStore { readonly #policy: EditabilityPolicy; + #revision = 0; + #snapshotRevision = 0; readonly #resources = new Map(); readonly #subscribers = new Set<() => void>(); @@ -76,7 +86,13 @@ export class LiveResourceStore { if (!existing) { // A resource we have never seen: base = draft = incoming. There is nothing to merge, and // nothing flashes — an arrival is not a change. - this.#resources.set(id, { server: incoming, draft: clone(incoming), redacted, conflicts: new Map() }); + this.#resources.set(id, { + revision: ++this.#revision, + server: incoming, + draft: clone(incoming), + redacted, + conflicts: new Map(), + }); this.#notify(); return { added: true, structural: true, flashed: [], conflicts: [] }; } @@ -94,6 +110,7 @@ export class LiveResourceStore { const structural = !sameShape(existing.draft, merged); // The REPLACEMENT (spec §4.1) and the base shift (I-BASESHIFT), in one line. Everything above // reconciled against the OLD server object; from here on, `incoming` is the base. + existing.revision = ++this.#revision; existing.server = incoming; existing.draft = merged; existing.redacted = redacted; @@ -116,6 +133,7 @@ export class LiveResourceStore { /** `reset`. Mark every known uid unseen — and prune NOTHING yet. */ beginSnapshot(): void { + this.#snapshotRevision++; this.#seen = new Set(); } @@ -149,7 +167,8 @@ export class LiveResourceStore { * You probably do not need this. The recommended save is 204: the write reaches the API server, the * watch echoes it back down the stream already projected, and the store converges. Dirty state is * derived from draft-versus-server, so there is nothing to adopt. Use this only when a host cannot - * wait for the echo. See docs/saving.md. */ + * wait for the echo. Unguarded adoption can overwrite newer watch state: for asynchronous + * responses use captureReconciliation before the request instead. See docs/saving.md. */ adoptSaved(object: KRMObject): void { const existing = this.#resources.get(object.metadata.uid); if (!existing) { @@ -309,6 +328,38 @@ export class LiveResourceStore { return out; } + /** Capture the patch, UID and merge-base version synchronously, before any await. A stale version + * is safe: Kubernetes rejects it with 409. Never replace it with a newer GET's version. */ + captureSave(id: string): SaveRequest | null { + const resource = this.#must(id); + const patch = this.patch(id); + if (!patch) return null; + const resourceVersion = resource.server.metadata.resourceVersion; + if (!resourceVersion) throw new Error("krm-stream: conditional save requires resourceVersion"); + return { uid: resource.server.metadata.uid, resourceVersion, patch }; + } + + /** Capture before starting a host GET. The returned function applies its projected response only + * if no server event or earlier response has advanced this resource since capture. Local edits + * remain valid and are three-way merged. Missing/recreated resources cannot be resurrected. + * Snapshot recovery also invalidates responses; a GET must never count as snapshot membership. + * The response must carry the same projection and redactions as the stream. */ + captureReconciliation(id: string): (object: KRMObject, opts?: ApplyOptions) => boolean { + const revision = this.#must(id).revision; + const snapshotRevision = this.#snapshotRevision; + return (object, opts = {}) => { + if ( + this.#seen !== null || + this.#snapshotRevision !== snapshotRevision || + object.metadata.uid !== id || + this.#resources.get(id)?.revision !== revision + ) + return false; + this.applyServerEvent(object, opts); + return true; + }; + } + /** A coarse "something changed" signal — the host re-renders and re-queries. Returns an * unsubscribe. */ subscribe(cb: () => void): () => void { diff --git a/packages/krm-stream/test/connection.test.ts b/packages/krm-stream/test/connection.test.ts new file mode 100644 index 0000000..54aaade --- /dev/null +++ b/packages/krm-stream/test/connection.test.ts @@ -0,0 +1,165 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { type ConnectionStatus, connectManagedResourceStream, LiveResourceStore } from "../src/index.ts"; + +const response = (events: unknown[]) => new Response(events.map((e) => `data: ${JSON.stringify(e)}\n\n`).join("")); +const object = (rv: string, value: string) => ({ + apiVersion: "v1", + kind: "ConfigMap", + metadata: { uid: "u", name: "cm", resourceVersion: rv }, + data: { value }, +}); + +test("managed recovery rejects a gap, resnapshots, and preserves a draft", async () => { + const store = new LiveResourceStore(); + store.applyServerEvent(object("1", "base")); + store.setValue("u", ["data", "value"], "draft"); + let calls = 0; + const states: ConnectionStatus[] = []; + const handle = connectManagedResourceStream("/stream", store, { + retryDelayMs: 0, + maxRetries: 1, + onStateChange: (s) => states.push(s.status), + fetch: async () => + ++calls === 1 + ? response([ + { seq: 1, type: "reset" }, + { seq: 3, type: "added", object: object("99", "must not apply") }, + ]) + : response([ + { seq: 1, type: "reset" }, + { seq: 2, type: "added", object: object("2", "base") }, + { seq: 3, type: "synced" }, + ]), + }); + await handle.closed; + assert.equal(calls, 2); + assert.equal(store.server("u").metadata.resourceVersion, "2"); + assert.deepEqual(store.draft("u").data, { value: "draft" }); + assert.ok(states.includes("retrying")); + assert.ok(states.includes("live")); + assert.equal(handle.state.status, "exhausted"); +}); + +for (const status of [401, 403]) + test(`HTTP ${status} is terminal`, async () => { + let calls = 0; + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + fetch: async () => { + calls++; + return new Response(null, { status }); + }, + retryDelayMs: 0, + }); + await handle.closed; + assert.equal(calls, 1); + assert.equal(handle.state.status, "terminal"); + }); + +test("terminal protocol errors never reconnect", async () => { + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + fetch: async () => response([{ seq: 1, type: "error", code: "FORBIDDEN", terminal: true }]), + }); + await handle.closed; + assert.equal(handle.state.status, "terminal"); + assert.equal(handle.state.retries, 0); +}); + +for (const failure of ["network", "http", "eof"]) + test(`${failure} consumes exactly the retry budget`, async () => { + let calls = 0; + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + maxRetries: 2, + retryDelayMs: 0, + fetch: async () => { + calls++; + if (failure === "network") throw new Error("offline"); + return new Response(null, { status: failure === "http" ? 503 : 200 }); + }, + }); + await handle.closed; + assert.equal(calls, 3); + assert.equal(handle.state.status, "exhausted"); + }); + +test("abort during backoff cancels the timer and never opens another connection", async () => { + const controller = new AbortController(); + let calls = 0; + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + signal: controller.signal, + fetch: async () => { + calls++; + throw new Error("offline"); + }, + onStateChange: (state) => { + if (state.status === "retrying") controller.abort(); + }, + }); + await handle.closed; + assert.equal(calls, 1); + assert.equal(handle.state.status, "closed"); +}); + +test("already aborted managed stream never fetches", async () => { + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + signal: AbortSignal.abort(), + fetch: async () => { + assert.fail("opened"); + }, + }); + await handle.closed; + assert.equal(handle.state.status, "closed"); +}); + +test("closing an active stream cancels the reader and prevents buffered events after close", async () => { + const store = new LiveResourceStore(); + let canceled = false; + let changes = 0; + const handle = connectManagedResourceStream("/stream", store, { + fetch: async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode('data: {"seq":1,"type":"reset"}\n\ndata: {"seq":2,"type":"synced"}\n\n'), + ); + }, + cancel() { + canceled = true; + }, + }), + ), + onChange: () => { + changes++; + handle.close(); + }, + }); + await handle.closed; + assert.equal(changes, 1); + assert.equal(canceled, true); + assert.equal(handle.state.status, "closed"); +}); + +test("abort cancels a quiet response reader", async () => { + let markOpen!: () => void; + const opened = new Promise((resolve) => { + markOpen = resolve; + }); + let canceled = false; + const handle = connectManagedResourceStream("/stream", new LiveResourceStore(), { + onOpen: markOpen, + fetch: async () => + new Response( + new ReadableStream({ + cancel() { + canceled = true; + }, + }), + ), + }); + await opened; + handle.close(); + await handle.closed; + assert.equal(canceled, true); + assert.equal(handle.state.status, "closed"); +}); diff --git a/packages/krm-stream/test/saving.test.ts b/packages/krm-stream/test/saving.test.ts new file mode 100644 index 0000000..603b3ed --- /dev/null +++ b/packages/krm-stream/test/saving.test.ts @@ -0,0 +1,120 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { LiveResourceStore } from "../src/index.ts"; + +const object = (rv: string, value = "base") => ({ + apiVersion: "v1", + kind: "ConfigMap", + metadata: { uid: "u", name: "cm", resourceVersion: rv }, + data: { value }, +}); + +test("save capture detaches patch and binds it to the merge base", () => { + const store = new LiveResourceStore(); + store.applyServerEvent(object("1")); + assert.equal(store.captureSave("u"), null); + store.setValue("u", ["data", "value"], "first"); + const save = store.captureSave("u"); + store.applyServerEvent(object("2")); + store.setValue("u", ["data", "value"], "second"); + assert.deepEqual(save, { uid: "u", resourceVersion: "1", patch: { data: { value: "first" } } }); +}); + +test("409 reconciliation preserves drafts and exposes conflicts", () => { + const store = new LiveResourceStore(); + store.applyServerEvent(object("1")); + store.setValue("u", ["data", "value"], "mine"); + const reconcile = store.captureReconciliation("u"); + store.setValue("u", ["data", "value"], "newer draft"); + assert.equal(reconcile(object("2", "theirs")), true); + assert.deepEqual(store.draft("u").data, { value: "newer draft" }); + assert.equal(store.conflicts("u").length, 1); + assert.equal(reconcile(object("2", "theirs")), false); +}); + +test("late reads and saves cannot overwrite a newer watch or resurrect deletions", () => { + const store = new LiveResourceStore(); + store.applyServerEvent(object("1")); + const read = store.captureReconciliation("u"); + const save = store.captureReconciliation("u"); + store.applyServerEvent(object("3", "watch")); + assert.equal(read(object("2")), false); + assert.equal(save(object("2")), false); + const deleted = store.captureReconciliation("u"); + store.removeResource("u"); + assert.equal(deleted(object("4")), false); + store.applyServerEvent(object("5")); + assert.equal(deleted(object("4")), false); +}); + +test("a reconciliation response for a replacement UID is rejected", () => { + const store = new LiveResourceStore(); + store.applyServerEvent(object("1")); + const reconcile = store.captureReconciliation("u"); + const replacement = object("2"); + replacement.metadata.uid = "replacement"; + assert.equal(reconcile(replacement), false); + assert.deepEqual(store.ids(), ["u"]); +}); + +test("array edits capture the replacement array with its original version", () => { + const store = new LiveResourceStore(); + store.applyServerEvent({ ...object("1"), spec: { items: ["a", "b"] } }); + store.setValue("u", ["spec", "items", 0], "mine"); + const captured = store.captureSave("u"); + store.applyServerEvent({ ...object("2"), spec: { items: ["a", "theirs"] } }); + assert.deepEqual(captured, { uid: "u", resourceVersion: "1", patch: { spec: { items: ["mine", "b"] } } }); +}); + +test("the complete editor reconciles an HTTP 409 and never retries the write blindly", async () => { + const { conditionalEditor } = await import("../../../examples/conditional-save/editor.ts"); + const store = new LiveResourceStore(); + store.applyServerEvent(object("1")); + store.setValue("u", ["data", "value"], "mine"); + const methods: string[] = []; + const editor = conditionalEditor(store, "u", "/save", async (_url, init) => { + methods.push(init?.method ?? "GET"); + if (init?.method === "PATCH") { + assert.deepEqual(JSON.parse(init.body as string), { + uid: "u", + resourceVersion: "1", + patch: { data: { value: "mine" } }, + }); + return new Response(null, { status: 409 }); + } + store.setValue("u", ["data", "value"], "typed while saving"); + return Response.json({ object: object("2", "theirs"), redacted: [] }); + }); + assert.equal(await editor.save(), "conflict"); + assert.deepEqual(methods, ["PATCH", "GET"]); + assert.deepEqual(store.draft("u").data, { value: "typed while saving" }); + assert.equal(store.conflicts("u").length, 1); + assert.equal(editor.saving, false); +}); + +test("the editor ignores a reconciliation GET overtaken by the watch", async () => { + const { conditionalEditor } = await import("../../../examples/conditional-save/editor.ts"); + const store = new LiveResourceStore(); + store.applyServerEvent(object("1")); + store.setValue("u", ["data", "value"], "mine"); + const editor = conditionalEditor(store, "u", "/save", async (_url, init) => { + if (init?.method === "PATCH") return new Response(null, { status: 409 }); + store.applyServerEvent(object("3", "watch")); + return Response.json({ object: object("2", "stale GET"), redacted: [] }); + }); + assert.equal(await editor.save(), "conflict"); + assert.deepEqual(store.server("u").data, { value: "watch" }); +}); + +test("a late GET cannot mark an object present in a recovery snapshot", () => { + const store = new LiveResourceStore(); + store.applyServerEvent(object("1")); + const beforeReset = store.captureReconciliation("u"); + store.beginSnapshot(); + const duringReset = store.captureReconciliation("u"); + assert.equal(beforeReset(object("2")), false); + assert.equal(duringReset(object("2")), false); + store.endSnapshot(); + assert.deepEqual(store.ids(), []); + assert.equal(beforeReset(object("2")), false); +}); diff --git a/packages/krm-stream/tsconfig.test.json b/packages/krm-stream/tsconfig.test.json index 64dc6d0..1666afb 100644 --- a/packages/krm-stream/tsconfig.test.json +++ b/packages/krm-stream/tsconfig.test.json @@ -8,9 +8,9 @@ "extends": "./tsconfig.json", "compilerOptions": { "noEmit": true, - "rootDir": ".", + "rootDir": "../..", // node:test / node:assert — the only reason @types/node is a devDependency. It ships nothing. "types": ["node"] }, - "include": ["src/**/*.ts", "test/**/*.ts", "e2e/**/*.ts"] + "include": ["src/**/*.ts", "test/**/*.ts", "e2e/**/*.ts", "../../examples/conditional-save/**/*.ts"] } From 8d5ed59939fca399024b5a1c040ce2676e8ba523 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Fri, 11 Sep 2026 08:21:16 +0000 Subject: [PATCH 2/9] fix: sustain stream recovery and preserve reconciliation redactions --- .github/dependabot.yml | 11 + README.md | 4 +- Taskfile.yml | 22 +- docs/saving.md | 11 +- docs/vue.md | 27 +- examples/README.md | 2 + examples/conditional-save/README.md | 7 +- examples/conditional-save/editor.ts | 7 +- examples/vanilla-browser/index.html | 14 +- .../vanilla-browser/tests/live-krm.spec.ts | 40 + examples/vue/README.md | 9 + examples/vue/package-lock.json | 695 ++++++++++++++++++ examples/vue/package.json | 13 + examples/vue/tsconfig.json | 5 + examples/vue/useLiveResource.test.ts | 64 ++ examples/vue/useLiveResource.ts | 29 + gateway/kube/authz_test.go | 28 +- .../kube/examples/conditionalsave/handler.go | 8 +- gateway/reauthorize.go | 9 + gateway/reauthorize_test.go | 55 ++ packages/krm-stream/README.md | 6 +- packages/krm-stream/src/connection.ts | 41 +- packages/krm-stream/src/index.ts | 2 +- packages/krm-stream/src/store.ts | 24 +- packages/krm-stream/test/connection.test.ts | 108 +++ packages/krm-stream/test/saving.test.ts | 34 +- 26 files changed, 1208 insertions(+), 67 deletions(-) create mode 100644 examples/vue/README.md create mode 100644 examples/vue/package-lock.json create mode 100644 examples/vue/package.json create mode 100644 examples/vue/tsconfig.json create mode 100644 examples/vue/useLiveResource.test.ts create mode 100644 examples/vue/useLiveResource.ts diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 17d0c4f..8bb5192 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -59,3 +59,14 @@ updates: patterns: ["*"] commit-message: prefix: "chore(deps)" + + # Vue and its test tools remain confined to the private adapter example. + - package-ecosystem: npm + directory: /examples/vue + schedule: + interval: weekly + groups: + npm: + patterns: ["*"] + commit-message: + prefix: "chore(deps)" diff --git a/README.md b/README.md index 6664453..cc84843 100644 --- a/README.md +++ b/README.md @@ -105,11 +105,11 @@ There are two halves, and they are usually two different people. No bundler, no framework, no Kubernetes client. `EventSource` is native, and the store is plain ESM: ```ts -import { LiveResourceStore, connectWithEventSource, resourceStreamURL } from "@configbutler/krm-stream"; +import { LiveResourceStore, connectManagedResourceStream, resourceStreamURL } from "@configbutler/krm-stream"; const store = new LiveResourceStore(); -connectWithEventSource( +connectManagedResourceStream( resourceStreamURL("/resource-stream/v1", { target: "production", version: "v1", diff --git a/Taskfile.yml b/Taskfile.yml index 6bb3ac8..e939781 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -91,7 +91,7 @@ tasks: # -------------------------------------------------------------------- test -- test: desc: "Run both suites against the shared conformance fixtures." - deps: [test-gateway, test-kube, test-client] + deps: [test-gateway, test-kube, test-client, test-vue] test-gateway: desc: "Go: the gateway + its half of the conformance suite." @@ -119,6 +119,24 @@ tasks: cmds: - node --test + test-vue: + desc: "Typecheck and test the optional Vue example." + deps: [_vue-deps] + dir: examples/vue + cmds: + - npm test + + _vue-deps: + internal: true + dir: examples/vue + sources: ["package.json", "package-lock.json"] + generates: ["node_modules/.package-lock.json"] + status: + - test -x node_modules/.bin/tsc + - test -f node_modules/vue/package.json + cmds: + - npm ci --no-audit --no-fund + # --------------------------------------------------------------------- e2e -- replay: desc: "Serve the conformance corpus over real SSE — a cluster you can point a browser at." @@ -163,7 +181,7 @@ tasks: # and it cannot do that with no node_modules there. On a developer's machine that directory is # already populated by `task test`, which is exactly why this gap survived until the first CI run # on a clean checkout found it. - deps: [fixtures, build-client, _example-deps] + deps: [fixtures, build-client, _example-deps, test-vue] dir: examples/vanilla-browser cmds: # The only place the library's central promise is actually tested: that the published ESM diff --git a/docs/saving.md b/docs/saving.md index 7ebc287..05431f0 100644 --- a/docs/saving.md +++ b/docs/saving.md @@ -27,8 +27,8 @@ before starting that GET: ```ts const reconcile = store.captureReconciliation(uid); -const { object, redacted } = await hostRead(); -reconcile(object, { redacted }); +const { object, redactedPaths } = await hostRead(); +reconcile(object, { redactedPaths }); // false means a newer server event/response won, or the UID disappeared/changed. Do not force it. ``` @@ -76,8 +76,11 @@ from overwriting a newer watch event, and preserves local edits made while savin `store.adoptSaved(object)` remains available for synchronous adoption and newly created objects. It is unguarded and must not receive delayed responses that can race the watch. Never adopt a raw Kubernetes response: project it first and provide the correct redaction metadata. Hosts exposing -redacted resources must keep read and stream redaction revisions consistent; the ConfigMap example -has no withheld values. +redacted resources can return `redactedPaths` directly from `gateway.Project`. The guard retains +known stream revisions for paths still present and removes paths absent from that list. Omitted +redaction metadata preserves existing protections. Unknown paths reject the entire response: open a +fresh stream snapshot before retrying reconciliation. Never invent revision counters for a GET. +An explicit `redacted` array is still supported when the host has authoritative stream revisions. ## Creating and deleting whole objects diff --git a/docs/vue.md b/docs/vue.md index b3ada01..53478b7 100644 --- a/docs/vue.md +++ b/docs/vue.md @@ -4,29 +4,12 @@ A copyable Vue 3 composable covers the adapter: reactive snapshots for drafts, c redactions and connection state, plus automatic subscription cleanup. Vue remains a host dependency; the core package and its browser bundle have no framework dependencies. -```ts -import { shallowRef, onScopeDispose } from "vue"; -import type { LiveResourceStore, ManagedStreamHandle } from "@configbutler/krm-stream"; +The [composable source](../examples/vue/useLiveResource.ts) and +[tests](../examples/vue/useLiveResource.test.ts) are typechecked and executed by `task test-vue` and CI. +Copy the source into your host and change its library import to `@configbutler/krm-stream`. -// Call in setup() or an active effect scope. One fixed UID per instance. -export function useLiveResource( - store: LiveResourceStore, - uid: string, - connection: ManagedStreamHandle, -) { - const read = () => store.ids().includes(uid) ? { - draft: store.draft(uid), - changes: store.changes(uid), - conflicts: store.conflicts(uid), - redactions: store.redactions(uid), - } : null; - const resource = shallowRef(read()); - const state = shallowRef(connection.state); - const stopStore = store.subscribe(() => { resource.value = read(); }); - const stopConnection = connection.subscribe(next => { state.value = next; }); - onScopeDispose(() => { stopStore(); stopConnection(); }); - return { resource, state }; -} +```ts +const { resource, state } = useLiveResource(store, uid, connection); ``` Use `resource.value?.draft` in script and `resource?.draft` in a template. A missing/deleted UID gives diff --git a/examples/README.md b/examples/README.md index 5cab45b..7c1725e 100644 --- a/examples/README.md +++ b/examples/README.md @@ -12,3 +12,5 @@ For host integration patterns, use these small recipes: [Conditional save](conditional-save/README.md) composes a managed connection with atomic save capture, 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. diff --git a/examples/conditional-save/README.md b/examples/conditional-save/README.md index 34bec18..2008780 100644 --- a/examples/conditional-save/README.md +++ b/examples/conditional-save/README.md @@ -27,7 +27,7 @@ connection.close(); ``` Use the same scope and `krm-full/v1` projection on the stream and this example endpoint. The GET -returns a complete projected object with redactions; it must use a most-recent Kubernetes read, with +returns a complete projected object with redacted paths; it must use a most-recent Kubernetes read, with no HTTP or application cache. A recreated name has a different UID and must be opened as a new editor. Keep an external draft archive if users need to recover edits after deletion: the store intentionally removes drafts of deleted objects. @@ -41,3 +41,8 @@ patch onto the latest version. No automatic write retry is performed. Successful writes return 204. The stream echo settles the saved values while retaining later edits. If the echo is delayed, dirty state remains visible; prevent repeated saves until your host's chosen acknowledgment UX allows them. Save results never feed raw Kubernetes objects back into the store. + +The GET returns `redactedPaths` from `gateway.Project`, without stream revision counters. The client +preserves revisions for known paths. Unknown redaction paths reject reconciliation; open a fresh +managed stream on the same store to obtain its snapshot before retrying. The editor returns +`conflict` without automatically retrying writes. Omitted redaction metadata never clears protection. diff --git a/examples/conditional-save/editor.ts b/examples/conditional-save/editor.ts index 11069bf..0f0af92 100644 --- a/examples/conditional-save/editor.ts +++ b/examples/conditional-save/editor.ts @@ -1,4 +1,4 @@ -import type { KRMObject, LiveResourceStore, Redaction } from "../../packages/krm-stream/src/index.ts"; +import type { KRMObject, LiveResourceStore } from "../../packages/krm-stream/src/index.ts"; /** Copy into the host. request is the host's fetch wrapper (session/CSRF/error policy). * Call save only while the managed connection is live. Keep the editor mounted during requests. */ @@ -10,6 +10,7 @@ export function conditionalEditor(store: LiveResourceStore, uid: string, url: st }, async save(): Promise<"unchanged" | "busy" | "saved" | "conflict"> { if (saving) return "busy"; + if (!store.ids().includes(uid)) return "conflict"; if (store.conflicts(uid).length) return "conflict"; const intent = store.captureSave(uid); // No await between patch and merge-base capture. if (!intent) return "unchanged"; @@ -28,8 +29,8 @@ export function conditionalEditor(store: LiveResourceStore, uid: string, url: st const latest = await request(url, { cache: "no-store" }); if (latest.status === 404) return "conflict"; // Let the stream remove the old UID. if (!latest.ok) throw new Error(`Reconciliation failed: HTTP ${latest.status}`); - const body = (await latest.json()) as { object: KRMObject; redacted: Redaction[] | null }; - reconcile(body.object, { redacted: body.redacted ?? [] }); + const body = (await latest.json()) as { object: KRMObject; redactedPaths: string[] }; + reconcile(body.object, { redactedPaths: body.redactedPaths }); // A newer watch/GET wins if reconcile returns false. Render the current store either way. // Ask the user to review conflicts and save again; capture a fresh intent on that click. return "conflict"; diff --git a/examples/vanilla-browser/index.html b/examples/vanilla-browser/index.html index 987ab8d..7ef553c 100644 --- a/examples/vanilla-browser/index.html +++ b/examples/vanilla-browser/index.html @@ -5,7 +5,7 @@ No framework. No bundler. No build step for THIS page. One