From 91ec4b9ebd61a6f8b4407453051797760a3a0351 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 14 Jul 2026 09:42:36 +0000 Subject: [PATCH] feat!: export gateway.Project, and return the whole StreamChange from applyStreamEvent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Items 2 and 3 from the gitops-api adoption report. Both are the same bug in two languages: the library computed what a host needed and then dropped it at the seam. # 2. The save path had no supported shape store.adoptSaved(object) is the obvious thing to reach for after a save, and there was no safe way to feed it: project() was unexported, so a host could either hand the browser a RAW object — managedFields, status, and the Secret values the projection exists to withhold, delivered through the one endpoint the stream does not guard — or skip adoptSaved entirely and let the watch echo settle it. The second one is right, and nothing said so. Now both halves are fixed: - saving.md states that 204 + let-the-watch-echo-it is the RECOMMENDED shape, and says why: the store converges on its own and dirty state is derived, so there is nothing to adopt. Its example previously ended with writeResult(w, result, err) — handing back the raw object, i.e. demonstrating the leak in the recommended code. It answers 204 now. - gateway.Project(projection, obj) is exported for hosts that must answer with the object, so adoptSaved can be fed something safe. - adoptSaved's own doc comment now says its argument MUST be projected. Project returns []string, not []Redaction: Rev is a counter the STREAM keeps per uid, incremented when a withheld value changes underneath a consumer who cannot see it. There is no honest Rev for a lone object, and inventing a zero would be a field that looks like an answer. # 3. applyStreamEvent threw away everything but the paths The store computes ApplyResult{added, structural, flashed, conflicts} and the seam returned .flashed alone, with no uid. So a UI rendering more than one resource per stream learned that something moved and not WHAT — and the only way out was to abandon connectWithEventSource, drive its own EventSource, and reimplement the event switch. That is the exact work this library exists to do once, on everyone's behalf. applyStreamEvent now returns StreamChange{type, uid, added, structural, flashed, conflicts}, and onChange receives it. Each field answers a question a renderer actually has: `added` distinguishes an arrival from a change, `structural` says rebuild-the-rows rather than re-read-the-values, and uid says which resource. A delete reports structural: true, because a row left. BREAKING CHANGE: applyStreamEvent returns StreamChange rather than Path[], and onChange receives a StreamChange rather than Path[]. Read `.flashed` for the previous value. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/saving.md | 39 +++++++++++++++- examples/vanilla-browser/index.html | 6 ++- gateway/patch_test.go | 52 +++++++++++++++++++++ gateway/project.go | 30 ++++++++++++ packages/krm-stream/e2e/wire.ts | 4 +- packages/krm-stream/src/index.ts | 2 +- packages/krm-stream/src/sse.ts | 61 +++++++++++++++++------- packages/krm-stream/src/store.ts | 11 +++++ packages/krm-stream/test/store.test.ts | 65 +++++++++++++++++++++++++- packages/krm-stream/test/wire.test.ts | 2 +- 10 files changed, 247 insertions(+), 25 deletions(-) diff --git a/docs/saving.md b/docs/saving.md index 34aa3ad..6266c4b 100644 --- a/docs/saving.md +++ b/docs/saving.md @@ -23,12 +23,47 @@ func (s *server) saveConfigMap(w http.ResponseWriter, r *http.Request) { } // Your app performs this ordinary Kubernetes PATCH as the signed-in user. - result, err := s.dynamicFor(user).Resource(configMaps).Namespace(scope.Namespace). + _, err := s.dynamicFor(user).Resource(configMaps).Namespace(scope.Namespace). Patch(r.Context(), scope.Name, types.MergePatchType, patch, metav1.PatchOptions{}) - writeResult(w, result, err) + if err != nil { + http.Error(w, "save failed", http.StatusBadGateway) + return + } + + // 204, and NOT the object Kubernetes just handed back. See below. + w.WriteHeader(http.StatusNoContent) } ``` +## Answer 204 and let the watch echo it + +This is the recommended shape, and it is the one to reach for unless you have a specific reason not +to. + +The object returned by a Kubernetes write is a *raw* object: `managedFields`, the last-applied +annotation, `status`, and the Secret values your projection withholds. Writing it to the response +hands the browser, through your save endpoint, precisely what the stream spent its whole design +refusing to send. The save endpoint is not covered by the projection unless you cover it. + +You do not need to. The write goes to the API server, the watch sees it, and it arrives back down the +stream as an ordinary `modified` event — projected, redacted, three-way merged into the draft the user +is still holding. The store converges on its own. Dirty state is derived from `draft` versus `server`, +so there is nothing to clear and nothing to adopt: the echo settles it. + +## If you must answer with the object + +`store.adoptSaved(object)` exists for a host that already holds a **projected** object — a host doing +its own optimistic update, or one that cannot wait a round-trip for the echo. 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) +``` + +`gateway.Project` applies the same projection the stream applies. Never hand `adoptSaved` an object +straight from the Kubernetes client. + The guard rejects: - a value declared in `redacted`, including deletion of a parent map such as `data: null`; diff --git a/examples/vanilla-browser/index.html b/examples/vanilla-browser/index.html index c93ad57..a3ce41c 100644 --- a/examples/vanilla-browser/index.html +++ b/examples/vanilla-browser/index.html @@ -92,8 +92,10 @@

patch — RFC 7386, editable changes only

let pending = new Set(); connectWithEventSource(`/resource-stream/v1?fixture=${fixture}&pace=${pace}`, store, { - onChange: (flashed) => { - for (const p of flashed) pending.add(key(p)); + // `change` names the resource (change.uid) as well as the paths, which is what a UI showing + // more than one object per stream needs — this demo shows one, so it only reads flashed. + onChange: (change) => { + for (const p of change.flashed) pending.add(key(p)); render(); }, onSynced: () => { statusLine.textContent = "synced — live"; }, diff --git a/gateway/patch_test.go b/gateway/patch_test.go index 1a1e0d5..8290a92 100644 --- a/gateway/patch_test.go +++ b/gateway/patch_test.go @@ -53,3 +53,55 @@ func TestValidateMergePatchHonorsProjectionAndPatchShape(t *testing.T) { } } } + +// Project is exported because the SAVE path needs it, and until it was, a host answering a save with +// the written object had no supported way to make that object safe. The obvious thing — hand back +// what the Kubernetes client returned — leaks exactly what the stream withholds, through the one +// endpoint the stream does not guard. +// +// So this asserts the property a host is relying on: what Project returns is safe to send to a +// browser, and it names what it withheld. +func TestProjectMakesASavedObjectSafeToReturn(t *testing.T) { + // What a Kubernetes write actually hands back. + raw := KRMObject{ + "apiVersion": "v1", + "kind": "Secret", + "metadata": map[string]any{ + "uid": "u-1", + "name": "creds", + "managedFields": []any{ + map[string]any{"manager": "kubectl"}, + }, + "annotations": map[string]any{ + lastAppliedAnnotation: `{"apiVersion":"v1"}`, + }, + }, + "data": map[string]any{"token": "s3cret"}, + } + + projected, redacted := Project(ProjectionFull, raw) + + meta, _ := projected["metadata"].(map[string]any) + if _, found := meta["managedFields"]; found { + t.Error("managedFields survived Project — a save response would put it in the browser") + } + if ann, found := meta["annotations"].(map[string]any); found { + if _, found := ann[lastAppliedAnnotation]; found { + t.Error("last-applied-configuration survived Project") + } + } + + data, _ := projected["data"].(map[string]any) + if got, found := data["token"]; found { + t.Errorf("the Secret VALUE survived Project: %v — this is the leak the projection exists to prevent", got) + } + if len(redacted) != 1 || redacted[0] != "/data/token" { + t.Errorf("redacted = %v, want [/data/token] — a consumer must still learn the key exists", redacted) + } + + // And the input is untouched: the caller's object may be an informer's, shared with every other + // stream on the same scope. + if raw["data"].(map[string]any)["token"] != "s3cret" { + t.Error("Project mutated its input") + } +} diff --git a/gateway/project.go b/gateway/project.go index 601ed80..35e33b9 100644 --- a/gateway/project.go +++ b/gateway/project.go @@ -47,6 +47,36 @@ type redactedValue struct { value any } +// Project applies a projection to an object, returning the object as it would go on the wire and the +// RFC 6901 pointers whose values were withheld from it. +// +// This exists because the SAVE path needs it. A host that answers a save with the written object must +// answer with a PROJECTED one — hand a browser what Kubernetes returned and you have just sent it +// managedFields and the Secret value that the whole projection exists to withhold, through the one +// endpoint the stream does not guard. Before this was exported there was no supported way to do that, +// which made `store.adoptSaved(object)` a trap: the obvious thing to reach for, and no safe way to +// feed it. +// +// The recommended save shape is still to answer 204 and let the watch echo the write back (see +// docs/saving.md): the store converges on its own, dirty state is derived, and nothing needs +// projecting. Use this when a host has a reason to hand the object back directly. +// +// # Why []string and not []Redaction +// +// Redaction carries a Rev, and Rev is not a property of an object — it is a counter the STREAM keeps +// per uid, incremented when a withheld value changes underneath a consumer that cannot see it. There +// is no honest Rev to give you here, and inventing a zero would be a field that looks like an answer. +// The paths are the part that is true of the object alone. `adoptSaved` needs no Rev either: it keeps +// the redactions the store already has. +func Project(p Projection, in KRMObject) (KRMObject, []string) { + out, values := project(p, in) + paths := make([]string, 0, len(values)) + for _, v := range values { + paths = append(paths, v.path) + } + return out, paths +} + func project(p Projection, in KRMObject) (KRMObject, []redactedValue) { out := deepCopyObject(in) redacted := []redactedValue{} diff --git a/packages/krm-stream/e2e/wire.ts b/packages/krm-stream/e2e/wire.ts index 997103b..ed228e0 100644 --- a/packages/krm-stream/e2e/wire.ts +++ b/packages/krm-stream/e2e/wire.ts @@ -89,8 +89,8 @@ for (const f of fixtures) { const connections = (f.watch ?? []).filter((op) => (op as { op: string }).op === "disconnect").length + 1; for (let conn = 0; conn < connections; conn++) { const handle = connectResourceStream(url(f.id, conn), store, { - onChange: (paths) => { - flashed.push(...paths); + onChange: (change) => { + flashed.push(...change.flashed); onEvent(); }, onError: () => onEvent(), // an `error` event occupies an index too (see resync-midstream) diff --git a/packages/krm-stream/src/index.ts b/packages/krm-stream/src/index.ts index 49adec5..7bb1467 100644 --- a/packages/krm-stream/src/index.ts +++ b/packages/krm-stream/src/index.ts @@ -21,7 +21,7 @@ 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 { StreamHandle, StreamOptions } from "./sse.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 { LiveResourceStore } from "./store.ts"; diff --git a/packages/krm-stream/src/sse.ts b/packages/krm-stream/src/sse.ts index 06543ee..dad1872 100644 --- a/packages/krm-stream/src/sse.ts +++ b/packages/krm-stream/src/sse.ts @@ -14,7 +14,7 @@ // be allowed to see — forever, from every open tab. import type { LiveResourceStore } from "./store.ts"; -import type { ErrorCode, Path, StreamEvent } from "./types.ts"; +import type { ErrorCode, EventType, Path, StreamEvent } from "./types.ts"; /** Incremental SSE parser. Bytes arrive in whatever chunks the network feels like — a frame can be * split down the middle, and it WILL be, under exactly the load where you least want to debug it — @@ -92,37 +92,66 @@ function parseFrame(frame: string): StreamEvent | null { * it is exported because it IS the protocol — a host feeding a store from its own transport should * not have to reimplement the switch and get `synced` subtly wrong. * - * Returns the paths that flashed, so a UI can highlight them. */ -export function applyStreamEvent(store: LiveResourceStore, ev: StreamEvent): Path[] { + * Returns the complete StreamChange — which resource, and what happened to it. */ +export function applyStreamEvent(store: LiveResourceStore, ev: StreamEvent): StreamChange { switch (ev.type) { case "reset": store.beginSnapshot(); - return []; + return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] }; case "added": - case "modified": - if (!ev.object) return []; - return store.applyServerEvent(ev.object, { redacted: ev.redacted }).flashed; - case "deleted": - if (ev.identity?.uid) store.removeResource(ev.identity.uid); - return []; + case "modified": { + if (!ev.object) return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] }; + const result = store.applyServerEvent(ev.object, { redacted: ev.redacted }); + return { type: ev.type, uid: ev.object.metadata.uid, ...result }; + } + case "deleted": { + const uid = ev.identity?.uid; + if (uid) store.removeResource(uid); + // A removal IS structural: a row left the collection, and a UI that only re-reads values would + // keep rendering it. + return { type: ev.type, uid, added: false, structural: true, flashed: [], conflicts: [] }; + } case "synced": store.endSnapshot(); - return []; + return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] }; default: // An unknown event type MUST be ignored, not treated as an error (spec §0). That is what lets // the gateway add an optional event type later without breaking a browser nobody can update. - return []; + return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] }; } } +/** What one stream event did to the store. + * + * This is the whole ApplyResult and the uid it belongs to, because anything less makes a host + * reimplement the stream loop to get the rest back. A UI rendering more than ONE resource per stream + * — which is most of them — cannot use a bare list of paths: it knows what moved and not what moved. + * + * Each field answers a question a renderer actually has: + * + * uid which resource. Absent only on reset/synced, which are about the stream, not an object. + * added an arrival, not a change. Animate it in; do not flash it as if a value moved. + * structural keys or rows appeared or disappeared. REBUILD the list; re-reading values is not enough. + * flashed the paths the server moved. Highlight these. + * conflicts the paths now conflicted, complete — not just the new ones. + */ +export interface StreamChange { + type: EventType; + uid?: string; + added: boolean; + structural: boolean; + flashed: Path[]; + conflicts: Path[]; +} + 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. */ 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; - /** Called after any change, with the paths that flashed. */ - onChange?: (flashed: Path[]) => void; + /** Called after any change, with what the event did and which resource it did it to. */ + onChange?: (change: StreamChange) => void; /** A missing or duplicated event was observed. The connection is closed; reconnect for a snapshot. */ onGap?: (expected: number, received: number) => void; /** Abort the stream from outside. */ @@ -238,8 +267,8 @@ function feed(store: LiveResourceStore, sequence: StreamSequence, ev: StreamEven opts.onError?.(ev.code ?? "INTERNAL", ev.message ?? "", ev.terminal ?? false); return ev.terminal === true; } - const flashed = applyStreamEvent(store, ev); + const change = applyStreamEvent(store, ev); if (ev.type === "synced") opts.onSynced?.(); - opts.onChange?.(flashed); + opts.onChange?.(change); return false; } diff --git a/packages/krm-stream/src/store.ts b/packages/krm-stream/src/store.ts index 85e5f88..b500691 100644 --- a/packages/krm-stream/src/store.ts +++ b/packages/krm-stream/src/store.ts @@ -139,6 +139,17 @@ export class LiveResourceStore { /** The save succeeded and this is the object it produced. The watch will echo it too — and that * echo is a harmless no-op (I-IDEMPOTENT) — but a UI should not have to wait for it to stop * showing the field as dirty. */ + /** Adopt the object a save returned. + * + * `object` MUST be projected — the same projection the stream uses. An object straight from a + * Kubernetes client carries managedFields, status and the Secret values the projection withholds, + * and handing it here puts all of them in the browser through the one endpoint the stream does not + * guard. Project it on the server with `gateway.Project` first. + * + * 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. */ adoptSaved(object: KRMObject): void { const existing = this.#resources.get(object.metadata.uid); if (!existing) { diff --git a/packages/krm-stream/test/store.test.ts b/packages/krm-stream/test/store.test.ts index b9282fd..c1fa125 100644 --- a/packages/krm-stream/test/store.test.ts +++ b/packages/krm-stream/test/store.test.ts @@ -27,7 +27,7 @@ for (const f of clientFixtures()) { // applyStreamEvent is library code, not test code: the switch from event to store call IS the // protocol (spec §4), and a host feeding a store from its own transport must not have to // reimplement it and get `synced` subtly wrong. - flashed.push(...applyStreamEvent(store, resolve(f, fe))); + flashed.push(...applyStreamEvent(store, resolve(f, fe)).flashed); for (const edit of f.client?.edits ?? []) { if (edit.after === i) applyEdit(store, edit); @@ -45,3 +45,66 @@ for (const f of clientFixtures()) { test("the client suite actually ran the corpus", () => { assert.ok(clientFixtures().length >= 10, "the client half of the corpus is missing fixtures"); }); + +// The gap that sent the first host to adopt this back to writing its own EventSource loop. +// +// applyStreamEvent used to return Path[] — the flashed paths and nothing else — which is unusable the +// moment a stream carries more than one resource: a UI learns that something moved, and not what. So +// a host that wanted to highlight per-resource had to abandon connectWithEventSource, drive its own +// EventSource, and reimplement the event switch. That is precisely the work this library exists to do +// once, correctly, on everyone's behalf. +// +// It returns the whole StreamChange now. This pins each field, because each one was being computed in +// the store and thrown away at the seam. +test("a change says WHICH resource changed, and what kind of change it was", () => { + const store = new LiveResourceStore(); + const object = { + apiVersion: "v1", + kind: "ConfigMap", + metadata: { uid: "u-1", name: "app", namespace: "default", resourceVersion: "1" }, + data: { greeting: "hello" }, + }; + + store.beginSnapshot(); + const arrival = applyStreamEvent(store, { seq: 1, type: "added", object }); + assert.equal(arrival.uid, "u-1", "an added event must name the resource it added"); + assert.equal(arrival.added, true, "an arrival is not a change to an existing object"); + + // A value moving is NOT structural: the keys are the same, so a UI re-reads rather than rebuilds. + const moved = applyStreamEvent(store, { + seq: 2, + type: "modified", + object: { ...object, metadata: { ...object.metadata, resourceVersion: "2" }, data: { greeting: "hi" } }, + }); + assert.equal(moved.uid, "u-1"); + assert.equal(moved.added, false, "a modification of a known uid is not an arrival"); + assert.equal(moved.structural, false, "no key appeared or disappeared"); + // metadata.resourceVersion flashes too — it is a read-only field and it genuinely moved. Assert on + // the one this test is about rather than pinning the whole set, which would be a test of the + // flashing rules and those have their own fixtures. + assert.ok( + moved.flashed.some((p) => p.join("/") === "data/greeting"), + `the changed value must flash: ${JSON.stringify(moved.flashed)}`, + ); + + // A key APPEARING is structural: a renderer that only re-reads known values never shows it. + const grew = applyStreamEvent(store, { + seq: 3, + type: "modified", + object: { + ...object, + metadata: { ...object.metadata, resourceVersion: "3" }, + data: { greeting: "hi", farewell: "bye" }, + }, + }); + assert.equal(grew.structural, true, "a new key must be reported as structural, or the row is never drawn"); + + // And a delete names the uid too — a host cannot remove a row it cannot identify. + const gone = applyStreamEvent(store, { + seq: 4, + type: "deleted", + identity: { uid: "u-1", apiVersion: "v1", kind: "ConfigMap", name: "app", namespace: "default" }, + }); + assert.equal(gone.uid, "u-1", "a deleted event must name the resource it removed"); + assert.equal(gone.structural, true, "a row left the collection"); +}); diff --git a/packages/krm-stream/test/wire.test.ts b/packages/krm-stream/test/wire.test.ts index 10b0c46..c008ad0 100644 --- a/packages/krm-stream/test/wire.test.ts +++ b/packages/krm-stream/test/wire.test.ts @@ -49,7 +49,7 @@ for (const f of wireFixtures()) { const store = new LiveResourceStore(); const flashed: Path[] = []; for (const [i, ev] of events.entries()) { - flashed.push(...applyStreamEvent(store, ev)); + flashed.push(...applyStreamEvent(store, ev).flashed); for (const edit of f.client?.edits ?? []) { if (edit.after === i) applyEdit(store, edit); }