From 9b092d6b957de022de8d05b5464e00ed228df670 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Sun, 4 Oct 2026 20:45:53 +0000 Subject: [PATCH 1/4] fix(client): never count an adopted save response as snapshot membership adoptSaved went through the stream's upsert path, which marks the UID as seen in an open snapshot cycle. A save response that landed after a reconnect's reset therefore kept the object alive through a snapshot that no longer contained it: a deleted UID survived an empty synced and nothing would ever remove it. Only the stream's own upserts count towards a snapshot now. adoptSaved and accepted guarded reads share a private update path that never marks membership; a response is still adopted, and synced prunes it if the snapshot does not send it. Co-Authored-By: Claude Opus 5.5 --- packages/krm-stream/src/store.ts | 21 +++++++++++++------ packages/krm-stream/test/saving.test.ts | 27 ++++++++++++++++++++++++- 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/packages/krm-stream/src/store.ts b/packages/krm-stream/src/store.ts index 82d07d0..1422813 100644 --- a/packages/krm-stream/src/store.ts +++ b/packages/krm-stream/src/store.ts @@ -126,8 +126,17 @@ export class LiveResourceStore { // ------------------------------------------------------------------ the stream in -- /** `added` and `modified` — the only two upsert spellings, and they are treated identically - * (spec §4). Both mean "here is this object's complete current state". */ + * (spec §4). Both mean "here is this object's complete current state", and during a snapshot cycle + * that the object is still in scope. */ applyServerEvent(object: KRMObject, opts: ApplyOptions = {}): ApplyResult { + this.#seen?.add(object.metadata.uid); + return this.#upsert(object, opts); + } + + /** Replace the server object and reconcile the draft. Membership is the caller's business: only the + * stream's own upserts count towards a snapshot, so a save or read response that lands mid-cycle + * can never keep alive an object the snapshot no longer contains. */ + #upsert(object: KRMObject, opts: ApplyOptions): ApplyResult { const id = object.metadata.uid; const incoming = clone(object); const redacted = (opts.redacted ?? []).map((entry) => ({ @@ -135,8 +144,6 @@ export class LiveResourceStore { rev: entry.rev, })); - this.#seen?.add(id); - const existing = this.#resources.get(id); if (!existing) { // A resource we have never seen: base = draft = incoming. There is nothing to merge, and @@ -222,14 +229,16 @@ export class LiveResourceStore { * 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 { + // Not snapshot membership: a response to a save that started before a resync says nothing about + // whether the object survived it. Only the snapshot does, and synced prunes what it did not send. const existing = this.#resources.get(object.metadata.uid); if (!existing) { - this.applyServerEvent(object); + this.#upsert(object, {}); return; } // A save response is another complete server object. Reconcile it rather than replacing the // draft so an edit made after the request was sent survives the response arriving. - this.applyServerEvent(object, { redacted: existing.redacted }); + this.#upsert(object, { redacted: existing.redacted }); } // ------------------------------------------------------------------------- edits -- @@ -417,7 +426,7 @@ export class LiveResourceStore { if (paths.some((path) => !known.has(pathKey(path)))) return false; redacted = paths.map((path) => known.get(pathKey(path))!); } - this.applyServerEvent(object, { redacted }); + this.#upsert(object, { redacted }); return true; }; } diff --git a/packages/krm-stream/test/saving.test.ts b/packages/krm-stream/test/saving.test.ts index 6c10201..54fbc87 100644 --- a/packages/krm-stream/test/saving.test.ts +++ b/packages/krm-stream/test/saving.test.ts @@ -1,6 +1,6 @@ import assert from "node:assert/strict"; import { test } from "node:test"; -import { LiveResourceStore } from "../src/index.ts"; +import { applyStreamEvent, LiveResourceStore } from "../src/index.ts"; const object = (rv: string, value = "base") => ({ apiVersion: "v1", @@ -131,6 +131,31 @@ test("a late GET cannot mark an object present in a recovery snapshot", () => { assert.equal(beforeReset(object("2")), false); }); +// The adopted-save ghost. A save response that lands after a reconnect's reset is not evidence that +// the object survived the disconnect: only the snapshot is. Counting it as seen kept a deleted UID +// alive through an empty snapshot, with nothing left that would ever remove it. +test("an adopted save response never counts as snapshot membership", () => { + const store = new LiveResourceStore(); + applyStreamEvent(store, { type: "reset" }); + applyStreamEvent(store, { type: "added", object: object("1") }); + applyStreamEvent(store, { type: "synced" }); + + applyStreamEvent(store, { type: "reset" }); + store.adoptSaved(object("2", "saved")); + assert.equal(store.server("u").metadata.resourceVersion, "2", "the response is still adopted"); + applyStreamEvent(store, { type: "synced" }); + assert.deepEqual(store.ids(), [], "deleted while disconnected: the empty snapshot prunes it"); + + // A create response mid-snapshot is held until synced, and kept only if the snapshot sends it. + const created = { ...object("3"), metadata: { uid: "new", name: "created", resourceVersion: "3" } }; + applyStreamEvent(store, { type: "reset" }); + store.adoptSaved(created); + assert.deepEqual(store.ids(), ["new"]); + applyStreamEvent(store, { type: "added", object: created }); + applyStreamEvent(store, { type: "synced" }); + assert.deepEqual(store.ids(), ["new"]); +}); + test("stateless reconciliation preserves redaction protection without inventing revisions", () => { const store = new LiveResourceStore(); const secret = { ...object("1"), kind: "Secret", data: {} }; From 40955b1cebdec6f7103c4698fa3a6211fe3beb94 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Sun, 4 Oct 2026 20:47:21 +0000 Subject: [PATCH 2/4] fix(client)!: stop the stream when any host callback throws Only the consumer's exceptions were handled precisely. A subscribe callback or onError that threw during a connection was swallowed as a network failure and retried until the budget ran out, so a buggy render function became a reconnect storm ending in exhausted. One thrown between connections rejected closed without publishing closed, and one thrown by the health timer's live escaped as an uncaught error. Every exception from host code now takes one path. The connector records the first, aborts the connection and stops without retrying, still notifies every subscriber of the state being published, ends in closed (unless it had already ended terminal or exhausted) and rejects closed with that exception after cleanup. The internal fetch loop no longer catches host code at all. The README says to attach a rejection handler to closed and not to pass an async consumer, which type-checks but can let live precede the applied synced and loses its exceptions. BREAKING CHANGE: an exception thrown by a subscribe callback or onError now stops the stream without retrying and rejects closed with that exception, as a consumer exception does. Co-Authored-By: Claude Opus 5.5 --- packages/krm-stream/README.md | 14 +- packages/krm-stream/src/connection.ts | 178 +++++++++++--------- packages/krm-stream/src/sse.ts | 51 +++--- packages/krm-stream/test/connection.test.ts | 97 +++++++++++ 4 files changed, 225 insertions(+), 115 deletions(-) diff --git a/packages/krm-stream/README.md b/packages/krm-stream/README.md index 387cfd5..9284791 100644 --- a/packages/krm-stream/README.md +++ b/packages/krm-stream/README.md @@ -70,7 +70,10 @@ connection.close(); bearer headers; `credentials: "include"` opts into cross-origin cookies. It decodes and sequence-checks the stream, and calls your callback synchronously, exactly once per state event (`reset`, `added`, `modified`, `deleted`, `synced`), in stream order, without the wire `seq`. -Errors go to `onError`, never to the callback; unknown event types are ignored. It requests a fresh +The callback must finish applying each event before it returns, so do not pass an `async` function: +it type-checks, but `live` could then be published before `synced` is applied, and its exceptions +would never reach `closed`. Errors go to `onError`, never to the callback; unknown event types are +ignored. It requests a fresh snapshot on sequence gaps, network failures, HTTP 408/429/5xx, EOF, and a connection the gateway closes after a retryable error such as `UPSTREAM_UNAVAILABLE`. Existing drafts survive recovery. @@ -88,9 +91,12 @@ errors (including 401/403, excluding 408/429) stop retries. `close()` or `signal pending backoff; `closed` resolves after cleanup. Create a new handle after credentials change or an explicit user retry. -If your callback throws, even after calling `close()`, the stream stops without retrying, publishes -`closed`, and `closed` rejects with that exception once the reader, timers and listeners are released. -Await `closed` or attach a rejection handler. A gateway's `X-KRM-Stream-Protocol` header is optional; +An exception from your own code is a bug, not a network failure, so it is never retried. If the +callback, a `subscribe` callback or `onError` throws, even after calling `close()`, the stream stops, +every subscriber still receives the state being published, the final state is `closed`, and `closed` +rejects with the first exception once the reader, timers and listeners are released. Always attach +a rejection handler (`connection.closed.catch(reportApplicationError)`) or await `closed`; otherwise +such a bug surfaces as an unhandled rejection. A gateway's `X-KRM-Stream-Protocol` header is optional; a stream that names a different protocol version is refused as a terminal `INTERNAL` error before any event is applied. diff --git a/packages/krm-stream/src/connection.ts b/packages/krm-stream/src/connection.ts index a353de0..c67db1b 100644 --- a/packages/krm-stream/src/connection.ts +++ b/packages/krm-stream/src/connection.ts @@ -16,7 +16,9 @@ export interface ConnectionState { /** Receives every resource state event exactly once, synchronously and in stream order. It must * finish applying the event before it returns: `live` is published only after it has applied - * `synced`. An exception it throws ends the stream; see ResourceStreamHandle.closed. */ + * `synced`. Do not pass an `async` function: it type-checks, because the result is ignored, but it + * returns before applying anything and its exceptions never reach `closed`. An exception it throws + * ends the stream; see ResourceStreamHandle.closed. */ export type ResourceEventConsumer = (event: ResourceStateEvent) => void; export interface ResourceStreamOptions { @@ -46,8 +48,9 @@ export interface ResourceStreamHandle { subscribe(callback: (state: Readonly) => void): () => void; close(): void; /** Settles once the stream has ended — closed, aborted, terminal or exhausted — and its reader, - * timers and listeners are released. It rejects, after the same clean-up and a `closed` state, with - * the consumer's own exception when the consumer threw, even if it called close() first. */ + * timers and listeners are released. It rejects, after the same clean-up, with the first exception + * your own code threw: the consumer, a `subscribe` callback or `onError`, even after calling + * close(). Attach a rejection handler, or a thrown render bug becomes an unhandled rejection. */ closed: Promise; } @@ -55,7 +58,13 @@ export interface ResourceStreamHandle { * each state event to `consume`. Every reconnect requests a fresh snapshot; whatever the consumer * holds survives it. HTTP 401/403, terminal protocol errors and a different protocol version stop * permanently. EOF, network failures, sequence gaps and retryable errors such as - * UPSTREAM_UNAVAILABLE consume a bounded retry budget. A consumer exception stops at once. + * UPSTREAM_UNAVAILABLE consume a bounded retry budget. + * + * An exception from the host's own code — the consumer, a `subscribe` callback or `onError` — is a + * bug, not a network failure: retrying would turn it into a reconnect storm. It stops the stream at + * once without a retry, every subscriber still sees the state being published, the final state is + * `closed` (unless the stream had already ended `terminal` or `exhausted`), and `closed` rejects with + * the first exception. * * A server's retry hint (an HTTP `Retry-After`, or an error event's `retryAfterMs`) sets the least * the next reconnect waits, within `maxRetryDelayMs`. A non-terminal event on an open connection does @@ -94,9 +103,24 @@ export function connectResourceStream( clearTimeout(healthTimer); healthTimer = undefined; }; + // The first exception the host's own code threw. Recorded wherever it happens, so a close() from + // inside the same callback cannot turn it into an ordinary ending. + let failure: { error: unknown } | undefined; + const fail = (error: unknown) => { + failure ??= { error }; + clearHealthTimer(); + controller.abort(); + }; + const call = (callback: () => void) => { + try { + callback(); + } catch (error) { + fail(error); + } + }; const publish = (status: ConnectionStatus, detail: Pick = {}) => { state = Object.freeze({ status, retries: state.retries, ...detail }); - for (const callback of subscribers) callback(state); + for (const callback of subscribers) call(() => callback(state)); }; const close = () => { clearHealthTimer(); @@ -105,88 +129,84 @@ export function connectResourceStream( opts.signal?.addEventListener("abort", close, { once: true }); if (opts.signal?.aborted) close(); + /** Connect and reconnect until the stream ends, and say how it ended. */ + const run = async (): Promise<"closed" | "terminal" | "exhausted"> => { + while (!controller.signal.aborted) { + publish("connecting"); + if (controller.signal.aborted) break; + hintMs = undefined; + let gap: ConnectionState["gap"]; + await streamOnce( + url, + { + consume: (event) => call(() => consume(event)), + opened: () => publish("syncing"), + reset: () => { + clearHealthTimer(); + publish("syncing"); + }, + synced: () => { + hintMs = undefined; + if (state.status !== "live") { + healthTimer = setTimeout(() => { + healthTimer = undefined; + if (!controller.signal.aborted && state.status === "live") { + state = { ...state, retries: 0 }; + publish("live"); + } + }, healthyResetMs); + } + publish("live"); + }, + gap: (expected, received) => { + clearHealthTimer(); + gap = Object.freeze({ expected, received }); + }, + error: (code, message, isTerminal, retryAfterMs) => { + if (isTerminal) clearHealthTimer(); + terminal ||= isTerminal; + if (!isTerminal && retryAfterMs !== undefined) hintMs = retryAfterMs; + call(() => opts.onError?.(code, message, isTerminal, retryAfterMs)); + }, + }, + controller.signal, + opts, + ); + clearHealthTimer(); + if (controller.signal.aborted) break; + if (terminal) return "terminal"; + if (state.retries >= maxRetries) return "exhausted"; + const ceiling = Math.min(cap, delay * 2 ** Math.min(state.retries, 30)); + const jittered = Math.floor(ceiling * (0.5 + Math.random() * 0.5)); + const wait = Math.min(cap, Math.max(jittered, hintMs ?? 0)); + state = { ...state, retries: state.retries + 1 }; + publish("retrying", { retryInMs: wait, ...(gap && { gap }) }); + 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(); + }); + } + return "closed"; + }; + // 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; - hintMs = undefined; - let gap: ConnectionState["gap"]; - const failure = await streamOnce( - url, - { - consume, - opened: () => publish("syncing"), - reset: () => { - clearHealthTimer(); - publish("syncing"); - }, - synced: () => { - hintMs = undefined; - if (state.status !== "live") { - healthTimer = setTimeout(() => { - healthTimer = undefined; - if (!controller.signal.aborted && state.status === "live") { - state = { ...state, retries: 0 }; - publish("live"); - } - }, healthyResetMs); - } - publish("live"); - }, - gap: (expected, received) => { - clearHealthTimer(); - gap = Object.freeze({ expected, received }); - }, - error: (code, message, isTerminal, retryAfterMs) => { - if (isTerminal) clearHealthTimer(); - terminal ||= isTerminal; - if (!isTerminal && retryAfterMs !== undefined) hintMs = retryAfterMs; - opts.onError?.(code, message, isTerminal, retryAfterMs); - }, - }, - controller.signal, - opts, - ); - clearHealthTimer(); - if (failure) { - // Not a Kubernetes error and not retryable: the application's own code failed. The reader is - // already released; report the end, then the exception itself. - publish("closed"); - throw failure.error; - } - 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 jittered = Math.floor(ceiling * (0.5 + Math.random() * 0.5)); - const wait = Math.min(cap, Math.max(jittered, hintMs ?? 0)); - state = { ...state, retries: state.retries + 1 }; - publish("retrying", { retryInMs: wait, ...(gap && { gap }) }); - 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"); + const ended = await run(); + // A host exception ends the stream as closed, whatever the loop was doing when it happened. + publish(failure ? "closed" : ended); } finally { clearHealthTimer(); opts.signal?.removeEventListener("abort", close); subscribers.clear(); } + if (failure) throw failure.error; }); return { close, diff --git a/packages/krm-stream/src/sse.ts b/packages/krm-stream/src/sse.ts index b2bcc6d..bc9817d 100644 --- a/packages/krm-stream/src/sse.ts +++ b/packages/krm-stream/src/sse.ts @@ -119,7 +119,8 @@ export function toStateEvent(wire: StreamEvent): ResourceStateEvent | null { } /** What one connection reports to connectResourceStream. A host observes the same moments through - * the handle's state, and the events themselves through its consumer. */ + * the handle's state, and the events themselves through its consumer. None of these throw: + * connectResourceStream catches the host's exceptions itself and aborts the signal instead. */ export interface ConnectionHooks { /** Receives each state event synchronously, in stream order. */ consume: (event: ResourceStateEvent) => void; @@ -143,24 +144,17 @@ export interface ConnectionOptions { headers?: Record; } -/** The consumer's exception, boxed so that even `throw undefined` cannot pass for a clean end. */ -export interface ConsumerFailure { - error: unknown; -} - /** Open one fetch connection and deliver its state events until it ends: EOF, a network failure, a - * refusal, a terminal error, a sequence gap, `signal`, or an exception from the consumer. + * refusal, a terminal error, a sequence gap or `signal`. * * It resolves in every case, after the reader and listeners are released, because whether to come - * back is the caller's decision. Only the consumer's own exception is returned. It is captured where - * it was thrown, so neither a close() from inside the consumer nor the clean-up after it can turn it - * into an ordinary ending. */ + * back is the caller's decision. */ export async function streamOnce( url: string, hooks: ConnectionHooks, signal: AbortSignal, opts: ConnectionOptions = {}, -): Promise { +): Promise { const controller = new AbortController(); const abort = () => controller.abort(); signal.addEventListener("abort", abort, { once: true }); @@ -168,7 +162,6 @@ export async function streamOnce( const sequence = new StreamSequence(); // Called detached, so a consumer never sees this hooks object as `this`. const { consume } = hooks; - let failure: ConsumerFailure | undefined; /** Check and deliver one decoded event; false when the connection must end. */ const deliver = (wire: StreamEvent): boolean => { @@ -190,19 +183,14 @@ export async function streamOnce( hooks.reset(); if (controller.signal.aborted) return false; } - try { - consume(event); - } catch (error) { - failure = { error }; - return false; - } - // A consumer that closed the stream while applying `synced` gets no later `live`. + consume(event); + // A consumer that closed the stream — or threw — while applying `synced` gets no later `live`. if (event.type === "synced" && !controller.signal.aborted) hooks.synced(); return true; }; try { - if (controller.signal.aborted) return undefined; + if (controller.signal.aborted) return; const res = await (opts.fetch ?? globalThis.fetch)(url, { signal: controller.signal, headers: { Accept: "text/event-stream", ...opts.headers }, @@ -221,26 +209,26 @@ export async function streamOnce( : "INTERNAL"; const terminal = res.status >= 400 && res.status < 500 && res.status !== 408 && res.status !== 429; const message = (await statusMessage(res, controller.signal)) ?? `stream: HTTP ${res.status}`; - if (controller.signal.aborted) return undefined; // closed while reading the refusal: report nothing + if (controller.signal.aborted) return; // closed while reading the refusal: report nothing hooks.error(code, message, terminal, retryAfter(res.headers.get("Retry-After"))); - return undefined; + return; } if (controller.signal.aborted) { await res.body.cancel().catch(() => {}); - return undefined; + return; } // The header is optional (spec §0), and a cross-origin response may hide it. Only a version // that is present and different is refused, before a single event of it is applied. const protocol = res.headers.get("X-KRM-Stream-Protocol"); if (protocol !== null && protocol.trim() !== String(PROTOCOL_VERSION)) { await res.body.cancel().catch(() => {}); - if (controller.signal.aborted) return undefined; + if (controller.signal.aborted) return; hooks.error( "INTERNAL", `stream: protocol mismatch: the gateway speaks X-KRM-Stream-Protocol ${JSON.stringify(protocol)}, this client speaks ${PROTOCOL_VERSION}`, true, ); - return undefined; + return; } hooks.opened(); @@ -254,13 +242,13 @@ export async function streamOnce( try { for (;;) { const { done, value } = await reader.read(); - if (done || controller.signal.aborted) return failure; + if (done || controller.signal.aborted) return; for (const ev of decoder.push(value)) { // One chunk can hold many events. Nothing after a close is delivered. - if (controller.signal.aborted) return failure; + if (controller.signal.aborted) return; if (!deliver(ev)) { - controller.abort(); // gap, terminal error or consumer failure: stop this connection - return failure; + controller.abort(); // a gap or a terminal error: stop this connection + return; } } } @@ -269,9 +257,8 @@ export async function streamOnce( await reader.cancel().catch(() => {}); } } catch { - // The network failed, or an observer threw. Either way this connection is over and the caller - // decides about the next. A consumer exception never lands here: deliver() captured it. - return failure; + // The network failed. This connection is over, and the caller decides about the next. Host + // callbacks never land here: connectResourceStream catches their exceptions itself. } finally { signal.removeEventListener("abort", abort); } diff --git a/packages/krm-stream/test/connection.test.ts b/packages/krm-stream/test/connection.test.ts index f3e0662..29e34da 100644 --- a/packages/krm-stream/test/connection.test.ts +++ b/packages/krm-stream/test/connection.test.ts @@ -329,6 +329,103 @@ test("even a thrown undefined is kept, not mistaken for a clean end", async () = assert.equal(handle.state.status, "closed"); }); +// A subscriber or onError that throws is the host's bug, exactly like a consumer that throws. Before, +// one thrown during a connection read as a network failure and retried until the budget ran out — a +// broken render function became a reconnect storm ending in `exhausted`. +test("a subscriber that throws during a connection stops the stream without retrying", async () => { + const boom = new Error("render failed"); + let calls = 0; + let body!: { cancelled: boolean }; + const handle = connectResourceStream("/stream", ignore, { + retryDelayMs: 0, + fetch: async () => { + calls++; + const opened = openBody( + sse([ + { seq: 1, type: "reset" }, + { seq: 2, type: "synced" }, + ]), + ); + body = opened.body; + return opened.response; + }, + }); + handle.subscribe((state) => { + if (state.status === "live") throw boom; + }); + const others = statuses(handle); + await assert.rejects(handle.closed, (error) => error === boom); + assert.equal(calls, 1, "a host bug is not retried"); + assert.equal(body.cancelled, true); + assert.deepEqual(others, ["connecting", "syncing", "syncing", "live", "closed"], "every subscriber still hears it"); + assert.equal(handle.state.status, "closed"); +}); + +test("a subscriber that throws between connections never opens another", async () => { + for (const at of ["connecting", "retrying"] as const) { + const boom = new Error(`threw on ${at}`); + let calls = 0; + const handle = connectResourceStream("/stream", ignore, { + retryDelayMs: 0, + fetch: async () => { + calls++; + throw new Error("offline"); + }, + }); + handle.subscribe((state) => { + if (state.status === at) throw boom; + }); + const states = statuses(handle); + await assert.rejects(handle.closed, (error) => error === boom); + assert.equal(calls, at === "connecting" ? 0 : 1, at); + assert.equal(states.at(-1), "closed", at); + } +}); + +test("a live republished by the health timer that throws ends the stream", async () => { + const boom = new Error("threw on the health reset"); + let body!: { cancelled: boolean }; + const handle = connectResourceStream("/stream", ignore, { + healthyResetMs: 5, + fetch: async () => { + const opened = openBody( + sse([ + { seq: 1, type: "reset" }, + { seq: 2, type: "synced" }, + ]), + ); + body = opened.body; + return opened.response; + }, + }); + let lives = 0; + handle.subscribe((state) => { + if (state.status === "live" && ++lives === 2) throw boom; + }); + await assert.rejects(handle.closed, (error) => error === boom); + assert.equal(body.cancelled, true); + assert.equal(handle.state.status, "closed"); +}); + +test("onError that throws stops the stream instead of retrying", async () => { + const boom = new Error("could not show the error"); + let calls = 0; + const handle = connectResourceStream("/stream", ignore, { + retryDelayMs: 0, + onError: () => { + throw boom; + }, + fetch: async () => { + calls++; + return new Response(null, { status: 503 }); + }, + }); + const states = statuses(handle); + await assert.rejects(handle.closed, (error) => error === boom); + assert.equal(calls, 1); + assert.deepEqual(states, ["connecting", "closed"]); +}); + test("a matching or absent protocol header is accepted", async () => { for (const headers of [{ "X-KRM-Stream-Protocol": "1" }, {}] as Record[]) { const seen: string[] = []; From 585baa1b9ccd1213229a10caebe5e1e334ee9cf5 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Sun, 4 Oct 2026 20:48:09 +0000 Subject: [PATCH 3/4] docs: add an upgrade guide from 0.7 One page maps every removed browser and gateway name to its replacement: the connector's callback, state and subscribe in place of its lifecycle callbacks, a first-snapshot promise without onSynced, and the nested StreamConfig literal. It also says to attach .catch to closed, not to pass an async consumer, and that a protocol mismatch is terminal. The READMEs link to it. Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + docs/migrating.md | 100 ++++++++++++++++++++++++++++++++++ packages/krm-stream/README.md | 3 +- 3 files changed, 103 insertions(+), 1 deletion(-) create mode 100644 docs/migrating.md diff --git a/README.md b/README.md index 1bf7682..c6fb117 100644 --- a/README.md +++ b/README.md @@ -207,6 +207,7 @@ browser. Use [`gateway.ValidateMergePatch`](gateway/patch.go) in the host save h - [Client state model](docs/client-state-model.md): drafts, conflicts, redactions, and keyed lists. - [Alternatives and prior art](docs/alternatives.md): how this differs from Kubernetes clients, browser dashboards, and config-as-data systems. - [Releasing](docs/releasing.md): release workflow and publication prerequisites. +- [Upgrading from 0.7](docs/migrating.md): the one browser connector and the gateway's shared stream configuration. ## Requirements and maturity diff --git a/docs/migrating.md b/docs/migrating.md new file mode 100644 index 0000000..86f839a --- /dev/null +++ b/docs/migrating.md @@ -0,0 +1,100 @@ +# Upgrading from 0.7 + +The release after 0.7 has one browser connector and one name for each gateway setting. Old names are +removed rather than kept as aliases, so the compiler or a failed import finds every call site. The +editor store's methods (`setValue`, `captureSave`, `captureReconciliation`, `adoptSaved` and so on) +are unchanged. + +## Browser client + +`connectResourceStream` is now the only connector. It takes a callback that receives each resource +state event, instead of a store: + +| 0.7 | Now | +|---|---| +| `connectManagedResourceStream(url, store, options)` | `connectResourceStream(url, event => applyStreamEvent(store, event), options)` | +| `connectResourceStream(url, store, options)` (one connection) | the same: there is one connector, with bounded retries | +| `connectWithEventSource(url, store, options)` | the same: fetch sends the session cookie | +| `onStateChange: state => …` | read `connection.state`, then `connection.subscribe(state => …)` | +| `onChange: change => …` | render from `applyStreamEvent`'s return value inside the callback | +| `onSynced: () => …` | a `synced` event in the callback, or the `live` state | +| `onOpen: () => …` | the `syncing` state | +| `onGap: (expected, received) => …` | `state.gap` on the `retrying` state that the gap caused | +| `ManagedStreamOptions`, `StreamOptions` | `ResourceStreamOptions` | +| `ManagedStreamHandle`, `StreamHandle` | `ResourceStreamHandle` | +| `onError` | unchanged | + +```js +const store = new LiveResourceStore(); +let firstSnapshot; +const loaded = new Promise((resolve) => (firstSnapshot = resolve)); + +const connection = connectResourceStream( + url, + (event) => { + render(applyStreamEvent(store, event)); + if (event.type === "synced") firstSnapshot(); // replaces onSynced + }, + { onError: showError }, +); +showConnection(connection.state); +const stopShowing = connection.subscribe(showConnection); // replaces onStateChange +connection.closed.catch(reportApplicationError); + +// On teardown: +stopShowing(); +connection.close(); +``` + +Three behaviors to account for: + +- **`closed` can reject.** If the callback, a `subscribe` callback or `onError` throws, the stream + stops without retrying and `closed` rejects with that exception. Attach `.catch` (or await it); + otherwise a rendering bug becomes an unhandled rejection. +- **The callback must be synchronous.** An `async` function type-checks, but it returns before + applying the event, so `live` can be published before the snapshot is applied, and its exceptions + never reach `closed`. +- **A protocol mismatch is terminal.** A stream whose `X-KRM-Stream-Protocol` header names another + version is refused with a terminal `INTERNAL` error. Keep a vendored client in step with the + gateway it talks to. + +`applyStreamEvent` takes the callback's events, which carry no `seq` and are never errors. A host +that feeds a store from its own transport passes the same shapes: `reset`, `added` or `modified` +with `object`, `deleted` with `identity`, and `synced`. + +## Go gateway + +`Options` and `Gateway` embed one `StreamConfig`, so the settings they share are written inside it: + +```go +gateway.Handler(gateway.Options{ + Principal: userFromSession, + Scopes: allowedScopes, + StreamConfig: gateway.StreamConfig{ + Authorizer: authorizeScope, + Clients: clientsForUser, + Projections: gateway.StaticProjection(gateway.ProjectionFull), + WriteTimeout: 10 * time.Second, + }, +}) +``` + +Assignments keep working through Go's field promotion: `options.Authorizer = …` and +`g.WriteTimeout = …` need no change. + +| 0.7 | Now | +|---|---| +| `gateway.Gateway{Auth: a}` | `gateway.Gateway{StreamConfig: gateway.StreamConfig{Authorizer: a}}` | +| `Projection: p` on `Options` or `Gateway` | `Projections: gateway.StaticProjection(p)`; nil means `ProjectionFull` | +| `g.ServeStream(w, r, principal, scope)` | `g.ServeStream(w, r, principal, scope, "")` | +| `g.ServeStreamProjection(w, r, principal, scope, requested)` | `g.ServeStream(w, r, principal, scope, requested)` | +| `g.Stream(ctx, principal, scope, sink)` | `g.Stream(ctx, principal, scope, "", sink)` | +| `g.StreamProjection(ctx, principal, scope, requested, sink)` | `g.Stream(ctx, principal, scope, requested, sink)` | +| `gateway.NewSharedBackend(upstream)` | `gateway.NewSharedBackend(upstream, gateway.SharedOptions{})` | +| `gateway.NewSharedBackendWithOptions(upstream, options)` | `gateway.NewSharedBackend(upstream, options)` | +| `gateway.HeartbeatInterval` | `StreamConfig.HeartbeatInterval`; zero means 20 seconds | +| `Scope.Query()` | the scope's fields in order: target, group, version, resource, namespace, name, labelSelector | + +The repository's test harness is no longer exported: `Fixture`, `WatchOp`, `FixtureEvent`, `Corpus`, +`LoadCorpus`, `LoadConformance`, `ScriptedBackend` and `NewScriptedBackend`. Neither is +`SSESink.Comment`. None of them is needed to serve a stream. diff --git a/packages/krm-stream/README.md b/packages/krm-stream/README.md index 9284791..47ae970 100644 --- a/packages/krm-stream/README.md +++ b/packages/krm-stream/README.md @@ -47,7 +47,8 @@ Chromium. Neither has a runtime dependency. ## Status Install `@configbutler/krm-stream`. This project is pre-1.0: the protocol and API may still change -before 1.0. Renamed APIs are removed rather than retained as compatibility aliases. See the repository [README](../../README.md), +before 1.0. Renamed APIs are removed rather than retained as compatibility aliases; see +[upgrading from 0.7](../../docs/migrating.md). See the repository [README](../../README.md), [client state model](../../docs/client-state-model.md), and [release guide](../../docs/releasing.md). ## Connections and conditional saves From 092165762af5ec9f11980a6fb483bdaab47c7157 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Sun, 4 Oct 2026 20:50:46 +0000 Subject: [PATCH 4/4] docs: slim proposal 0009's change 2 and defer the read-only store Review found change 2 bundled removals, the snapshot tracker, a new ResourceStore, entry points and new option shapes, and that the case for ResourceStore is thin: its reference use is the repository's own example, and krm-foyer, the one known consumer, edits. - Change 2 is now the editor removals and fixes only. - The adopted-save membership bug ships ahead of it as a fix:. - SnapshotTracker, ResourceStore and the read-only example, and the stream/editor entry points, are deferred with explicit triggers: a read-only view that should not load the merge engine, or an unbundled consumer that benefits from loading only the stream layer. - Exceptions from subscribe callbacks and onError follow the consumer's rule (stop, no retry, reject closed), replacing the earlier out-of-scope note, and async consumers are documented as unsupported. - Proposal 0010 records its implementation, the nested-literal cost and the one golden message line it changes. Both proposals are re-wrapped at 100 columns. The previous reflow left fragments such as `state` and `}` on lines of their own, one of them splitting an inline code span; only whitespace changes there. Co-Authored-By: Claude Opus 5.5 --- .../0009-stream-and-editor-separation.md | 284 ++++++++++-------- docs/proposals/0010-gateway-api-cleanup.md | 69 +++-- 2 files changed, 187 insertions(+), 166 deletions(-) diff --git a/docs/proposals/0009-stream-and-editor-separation.md b/docs/proposals/0009-stream-and-editor-separation.md index fdee31c..0ee0396 100644 --- a/docs/proposals/0009-stream-and-editor-separation.md +++ b/docs/proposals/0009-stream-and-editor-separation.md @@ -1,16 +1,19 @@ # Proposal 0009 Stream and editor separation -**Status: proposed implementation plan.** The APIs below are proposed; they are not available yet. -Source review baseline: `a8281c5`, 2026-10-04. +**Status: in progress.** Change 1, one connector that delivers state events, is implemented in +[#53](https://github.com/ConfigButler/krm-stream/pull/53); the host-callback exception rule and the +adopted-save `fix:` follow on the same stack. Change 2 is proposed. The read-only store and the +stream/editor entry points are [deferred](#deferred-until-a-read-only-view-needs-them) until a +read-only view needs them. Source review baseline: `a8281c5`, 2026-10-04. krm-stream should make live Kubernetes resources easy to consume, with editing available as an additional layer. Keep one repository, one npm package and the existing lockstep release process. Separate browser transport and authoritative resource state from local drafts, three-way reconciliation, conflicts and patch capture. The Go gateway continues to serve the same v1 protocol. -Follow the [design rules](../../CONTRIBUTING.md#design-rules) and -[release policy](../releasing.md). This proposal owns the separation; existing behavioral follow-ups -remain in [proposal 0006](0006-stream-and-save-implementation-plan.md). +Follow the [design rules](../../CONTRIBUTING.md#design-rules) and [release policy](../releasing.md). +This proposal owns the separation; existing behavioral follow-ups remain in [proposal +0006](0006-stream-and-save-implementation-plan.md). ## Why live watches and three-way merge belong together @@ -60,8 +63,8 @@ compared three ways: If only theirs differs from base, the draft follows the server and the UI can highlight the change. If only ours differs, the edit stays. If both changed it to the same value, they have converged. If both changed it differently, the edit stays and the field is recorded as a conflict. The incoming -object then becomes `server`, and so the next base. The -[client state model](../client-state-model.md#editing-and-conflicts) has the full table. +object then becomes `server`, and so the next base. The [client state +model](../client-state-model.md#editing-and-conflicts) has the full table. ```mermaid sequenceDiagram @@ -92,16 +95,15 @@ identity, projection and recovery contract: - A captured save binds the person's patch to the identity and version they reviewed. Projection and redaction determine what the person may see and edit. Snapshot pruning can remove a -deleted resource's draft, and merge patches replace arrays as whole values. The -[client state model](../client-state-model.md) and [saving guide](../saving.md) explain those -limits. +deleted resource's draft, and merge patches replace arrays as whole values. The [client state +model](../client-state-model.md) and [saving guide](../saving.md) explain those limits. The live editor consumes the protocol's state; transport does not depend on the editor. The editor can also reconcile directly supplied server objects without opening a connection. The current connectors do not show that boundary: a list page must construct an editing store to use them. This proposal makes transport deliver protocol events, adds a small read-only store for lists and -viewers, -and keeps the editor as the layer a live form adds. Both stores use the same stream contract. +viewers, and keeps the editor as the layer a live form adds. Both stores use the same stream +contract. ## Current coupling and scope @@ -137,8 +139,9 @@ flowchart LR A view chooses one active store. An editor keeps its own authoritative base; it does not mirror a `ResourceStore`. Both stores use the same internal snapshot tracker, with a separate instance per -store. Share small value and dispatch helpers where useful. Keep framework adapters and a generic -merge library outside this work. +store. Until a read-only view needs `ResourceStore`, the editor is the only store, and a list +renders from a `LiveResourceStore` with a read-only policy. Share small value and dispatch helpers +where useful. Keep framework adapters and a generic merge library outside this work. ## One public connector @@ -149,8 +152,8 @@ Fetch already supports both same-origin cookies and bearer headers; a second pub unnecessary for either authentication case. The native EventSource helper closes on a sequence gap without arranging a new snapshot connection, -and leaves reconnect timing to the browser instead of enforcing the protocol's retry hints. This -is a limitation of that helper, not a reason to remove native SSE support from the gateway. +and leaves reconnect timing to the browser instead of enforcing the protocol's retry hints. This is +a limitation of that helper, not a reason to remove native SSE support from the gateway. The proposed public signature is: @@ -167,66 +170,64 @@ function connectResourceStream( `ResourceStreamHandle` keeps `state`, `subscribe`, `close` and `closed`. `ResourceStreamOptions` keeps `onError`, `signal`, injected fetch, credentials, headers and the existing retry settings. Remove public `onOpen`, `onSynced`, `onGap`, `onStateChange` and `onChange`. Observe connection -state -through `state` plus `subscribe`, and errors through `onError`. Store results or subscriptions drive -rendering. A consumer already receives `synced` when it needs per-snapshot work. +state through `state` plus `subscribe`, and errors through `onError`. Store results or subscriptions +drive rendering. A consumer already receives `synced` when it needs per-snapshot work. Keep the existing statuses: `connecting`, `syncing`, `live`, `retrying`, `closed`, `terminal` and `exhausted`. Defer opening until the handle exists so subscribers can observe the initial -transition. -Read `state` for initial presentation; `subscribe` observes subsequent publications. Preserve a -`live` publication after every completed snapshot, including when the previous status was live. -A sequence gap must remain diagnosable: add an optional `gap: { expected: number; received: number -}` -to the `retrying` state caused by that gap, clearing it on the next attempt. Keep other diagnostics -in `onError`; do not add another lifecycle callback or a general retry telemetry API. +transition. Read `state` for initial presentation; `subscribe` observes subsequent publications. +Preserve a `live` publication after every completed snapshot, including when the previous status was +live. A sequence gap must remain diagnosable: add an optional +`gap: { expected: number; received: number }` to the `retrying` state caused by that gap, clearing +it on the next attempt. Keep other diagnostics in `onError`; do not add another lifecycle callback +or a general retry telemetry API. ### Delivery and completion -1. Decode and check the per-connection sequence before delivering state. On a gap, discard the - event beyond the gap, close the connection and request a fresh snapshot through bounded retry. +1. Decode and check the per-connection sequence before delivering state. On a gap, discard the event + beyond the gap, close the connection and request a fresh snapshot through bounded retry. 2. Observe `reset` directly in transport: enter `syncing` and clear the healthy-period timer before state application. This must not depend on what a store returns. 3. Deliver recognized state events exactly once and synchronously, in stream order. Unknown wire event types still participate in sequence accounting and are then ignored. Error events go only to `onError`. Discard an upsert or deletion without its required object or identity; this - preserves - current no-op state behavior and is not comprehensive object schema validation. + preserves current no-op state behavior and is not comprehensive object schema validation. 4. Apply `synced` through the consumer before publishing `live` and starting the healthy-period timer. Pruning already precedes `live` today; rendering moves from the old `onChange`, which ran afterwards, into state application before `live`. -5. Preserve retry hints, exhaustion, terminal refusal and in-band `RESYNC_REQUIRED` recovery. - A retryable error does not itself restart an open connection; reconnect when that connection - ends. +5. Preserve retry hints, exhaustion, terminal refusal and in-band `RESYNC_REQUIRED` recovery. A + retryable error does not itself restart an open connection; reconnect when that connection ends. 6. After close or abort, deliver no later events from the same decoded chunk. If the consumer closes during `synced`, publish no later `live` transition. The existing fetch loop already checks abort between events; preserve this and test it. -The consumer must complete state application synchronously. Both stores expose `applyStreamEvent` +The consumer must complete state application synchronously. An `async` consumer type-checks, because +its result is ignored, but returns before applying anything and its exceptions never reach `closed`; +the documentation says so rather than adding a runtime check. Both stores expose `applyStreamEvent` as a bound arrow field, so passing `store.applyStreamEvent` directly is safe and preserves `this`. Verify direct method passing as well as wrapped consumers that render the returned change. -If a consumer throws, retain that exception even when the consumer first calls `close()`. The -current -fetch loop ignores errors after abort and swallows its completion rejection; neither rule may hide a -consumer exception. Capture it at invocation, finish reader/listener/timer cleanup, publish `closed` -and reject `closed` with the original exception. Do not retry it or classify it as a Kubernetes -error. -Ordinary network failure still follows the existing retry policy. Callers await `closed` or attach a -rejection handler; tests explicitly observe failure, including close-then-throw. Completion settles -only after the consumer returns and cleanup finishes. This does not authorize redesigning exceptions -from unrelated observer callbacks. +An exception from host code is a bug, not a network failure, so it is never retried. The rule covers +the consumer, `subscribe` callbacks and `onError` alike. Review found the first implementation +handled only the consumer: an observer that threw during a connection was swallowed as a network +failure and retried until the budget ran out, so a render bug became a reconnect storm, and one that +threw between connections rejected `closed` without publishing `closed`. Capture the first exception +where it is thrown, even after the callback called `close()`. Finish reader/listener/timer cleanup, +still notify every subscriber of the state being published, publish `closed` unless the stream +already ended `terminal` or `exhausted`, and reject `closed` with that exception. Do not classify it +as a Kubernetes error. Ordinary network failure still follows the existing retry policy. Callers +await `closed` or attach a rejection handler; tests explicitly observe failure, including +close-then-throw. Completion settles only after the callback returns and cleanup finishes. ### Protocol version diagnostics -On a successful streaming response, inspect `X-KRM-Stream-Protocol` before consuming events. -A present value other than the supported version is a terminal local `INTERNAL` error with a clear +On a successful streaming response, inspect `X-KRM-Stream-Protocol` before consuming events. A +present value other than the supported version is a terminal local `INTERNAL` error with a clear protocol-mismatch message: cancel the response body, publish `terminal` and do not retry. A missing header is accepted because [v1 versioning](../../spec/v1.md#0-versioning) makes this diagnostic -header -optional. HTTP refusals retain their existing classification. No version negotiation or new wire -error is introduced. Keep the protocol constant internal and its Go/TypeScript consistency test. -Keep public `VERSION`: build provenance has an established use for vendored assets. +header optional. HTTP refusals retain their existing classification. No version negotiation or new +wire error is introduced. Keep the protocol constant internal and its Go/TypeScript consistency +test. Keep public `VERSION`: build provenance has an established use for vendored assets. ## One public stream input per store @@ -238,14 +239,17 @@ passes state events; hosts can supply the same state shapes directly without ope Both stores expose a bound `applyStreamEvent(event)` method. In `LiveResourceStore`, make `beginSnapshot`, `endSnapshot`, `applyServerEvent` and `removeResource` private. Remove the -standalone -`applyStreamEvent(store, event)` export and fold public `ApplyResult` into `StreamChange`. Remove -`ApplyOptions`: direct state input accepts only the wire redaction shape, while stored/read -redactions -may still use parsed path segments. Tests must use the public event input for behavior; private -helpers remain available for the editor's guarded reconciliation implementation. +standalone `applyStreamEvent(store, event)` export and fold public `ApplyResult` into +`StreamChange`. Remove `ApplyOptions`: direct state input accepts only the wire redaction shape, +while stored/read redactions may still use parsed path segments. Tests must use the public event +input for behavior; private helpers remain available for the editor's guarded reconciliation +implementation. -### Shared snapshot tracking +### Shared snapshot tracking (deferred) + +Deferred with `ResourceStore`: extract the tracker when a second store needs it. Until then the +editor keeps its private membership state, where the membership rule below already holds: only +stream upserts mark a UID as seen. Extract `#seen` and `#snapshotRevision` into one internal `SnapshotTracker` before adding `ResourceStore`. It owns only the active membership pass, seen UIDs and generation: @@ -262,14 +266,16 @@ and UID checks in the editor. Use both stores to verify interrupted and restarte editor to verify late reads before, during and after a recovery cycle. Keep tracker internals private. -### Read-only store +### Read-only store (deferred) + +Deferred until a read-only view needs it. By this proposal's own account the reference use is the +repository's example, not an external request, and krm-foyer, the one known consumer, edits. The +design stays here so the store can be built as its own change when a list or viewer needs it. Add a read-only list to the [vanilla browser example](../../examples/vanilla-browser/README.md), -which -currently demonstrates editing. Show resource identity, status and redaction metadata without edit -controls. This is the concrete reference use case for `ResourceStore`; it is not an asserted -external -consumer request. Its public surface is: +which currently demonstrates editing. Show resource identity, status and redaction metadata without +edit controls. This is the concrete reference use case for `ResourceStore`; it is not an asserted +external consumer request. Its public surface is: ```ts class ResourceStore { @@ -291,16 +297,14 @@ interface ResourceChange { Replace complete projected objects, never deep-merge them. Reset removes nothing, upserts mark seen, and only synced prunes. Replayed events are idempotent. A new UID under the same name starts a separate entry. Copy incoming objects/redactions and return detached reads using existing -missing-UID -conventions. Notify after object or membership changes; connection state comes from the handle. -Each store belongs to one target and scope. Independent streams must use independent store +missing-UID conventions. Notify after object or membership changes; connection state comes from the +handle. Each store belongs to one target and scope. Independent streams must use independent store instances. `LiveResourceStore(readOnlyPolicy)` still retains a draft and runs reconciliation. Remove the named `readOnlyPolicy` definition/export once `ResourceStore` exists. Preserve all-read-only editor policy coverage through `regionPolicy([])`; a read-only editor policy and a resource viewer remain -different -uses. `ResourceStore` loads no merge engine and exposes no editing API. +different uses. `ResourceStore` loads no merge engine and exposes no editing API. ```ts import { ResourceStore, connectResourceStream } from "@configbutler/krm-stream/stream"; @@ -335,36 +339,32 @@ removal. The editor reports `structural: true` on snapshot pruning that removes its current upsert/highlight/conflict results. Define common change types independently of either store so the editor does not import `ResourceStore`. -Remove `adoptSaved`. It is unguarded and also marks membership during recovery. A baseline -reproduction -shows an old save response adopted after reset survives an empty synced snapshot, retaining a -deleted -UID. Add regression evidence while removing the unsafe input, rather than preserving that behavior -in the extracted tracker. Remove public optimistic `removeResource` alongside the other direct -inputs. +Remove `adoptSaved`: it is unguarded, so an old save response can still overwrite newer watch state. +It also marked membership during recovery: an old save response adopted after reset survived an +empty synced snapshot, retaining a deleted UID. That half is a bug, fixed separately as a `fix:` +ahead of change 2 with a regression test, so the removal no longer carries a correctness fix. Remove +public optimistic `removeResource` alongside the other direct inputs. Successful edits normally return 204 and converge through the watch echo. For an existing UID, a -host -returning an object uses `captureReconciliation(uid)` captured before the request. The guard refuses -newer bases, deleted/recreated UIDs and snapshot overlap; it does not insert a new resource. Creates -wait for their added echo. A host tracks successful creates as pending confirmation so they cannot -be submitted twice, correlating an echo with server-assigned identity/receipt when available. Handle -an echo arriving before the create response as well as after it. Pending deletes likewise wait for -stream absence, rather than removing authoritative state optimistically. Snapshot completion can -confirm missed echoes. Host write status and host confirmation state are separate from store data. -Use a small host-state recipe and executed tests; this does not add create/delete endpoints to the -library or turn the replay demo into a CRUD backend. +host returning an object uses `captureReconciliation(uid)` captured before the request. The guard +refuses newer bases, deleted/recreated UIDs and snapshot overlap; it does not insert a new resource. +Creates wait for their added echo. A host tracks successful creates as pending confirmation so they +cannot be submitted twice, correlating an echo with server-assigned identity/receipt when available. +Handle an echo arriving before the create response as well as after it. Pending deletes likewise +wait for stream absence, rather than removing authoritative state optimistically. Snapshot +completion can confirm missed echoes. Host write status and host confirmation state are separate +from store data. Use a small host-state recipe and executed tests; this does not add create/delete +endpoints to the library or turn the replay demo into a CRUD backend. Simplify `ReconciliationOptions` to `{ redactedPaths?: string[] }`. Omission preserves current protections; an explicit list removes absent paths, preserves known revisions and rejects unknown paths rather than inventing revision counters. Remove the `redacted` response branch from the conditional editor example and revise its tests. Stream input still carries wire redaction -revisions; -a stateless GET does not. Preserve edits made while a save/read was pending. +revisions; a stateless GET does not. Preserve edits made while a save/read was pending. Remove `takeTheirs`; `revert` already implements that operation. Remove the unused `status(uid)` -convenience method; read `server(uid).status`. Update the keep-local recipe in -[proposal 0006](0006-stream-and-save-implementation-plan.md) to use `revert`. +convenience method; read `server(uid).status`. Update the keep-local recipe in [proposal +0006](0006-stream-and-save-implementation-plan.md) to use `revert`. ```ts import { connectResourceStream } from "@configbutler/krm-stream/stream"; @@ -383,6 +383,11 @@ const intent = editor.captureSave(uid); ## Package boundaries and public exports +The `stream` and `editor` subpaths are deferred with `ResourceStore`: they help an unbundled browser +that loads only the stream layer, and krm-foyer's example vendors the single-file bundle, which +contains both layers anyway. Change 2 still trims the root exports as described at the end of this +section. + | Import | Exports | |---|---| | `@configbutler/krm-stream/stream` | Connector, lifecycle types, resource state/wire types, URL builder, `ResourceStore`, version stamp | @@ -401,43 +406,52 @@ Remove public exports of `clone`, `deepEqual`, `get`, `has`, `isPrefix`, `parseP Retain useful policies, schema helpers and public domain types. Verify repository and available host imports before implementing removals; unavailable host source is not evidence of no users. Add `sideEffects: false` only after auditing top-level effects; it can assist bundlers but does not -replace -explicit entry boundaries. Keep one npm package, zero runtime dependencies and existing ESM +replace explicit entry boundaries. Keep one npm package, zero runtime dependencies and existing ESM emission. ## Implementation order and migration -Deliver three reviewable changes, keeping each intermediate commit buildable: +Deliver in this order, keeping each intermediate commit buildable. Review split the original change +2, which bundled removals, the tracker, a new store and entry points; only the removals and fixes +remain planned. | Change | Deliver | Acceptance evidence | |---|---|---| -| 1 | One managed connector, event consumers, consolidated lifecycle observation, consumer exceptions and version diagnostics | Plain callbacks recover on gaps, respect retry policy and expose exceptions after cleanup; all connector callers migrated | -| 2 | Store methods/state types, shared tracker, read-only store, guarded save-only reconciliation, smaller exports and entry points | Both stores converge; late-save/snapshot regression passes; published stream entry loads without editor code | -| 3 | Read-only example, revised editor/create/delete guidance, metadata and protocol wording | Browser and Go-wire checks exercise both stores and raw native SSE compatibility; docs match the public API | +| 1 | One connector, event consumers, consolidated lifecycle observation, host-callback exceptions, version diagnostics, spec §7 and consumer rule 9 wording | Plain callbacks recover on gaps, respect retry policy and expose host exceptions after cleanup without retrying; all connector callers migrated; raw EventSource browser check kept | +| Fix | Adopted save responses never count as snapshot membership, as a `fix:` | A response adopted after a reset does not survive an empty snapshot | +| 2 | Editor removals and fixes: bound `applyStreamEvent` with private snapshot and upsert methods; the standalone helper, `ApplyResult`, `ApplyOptions`, `adoptSaved`, `takeTheirs` and `status()` removed; simplified `ReconciliationOptions`; trimmed root exports; create/delete host guidance | Editor behavior through the public event input; late-read and snapshot regressions; host examples confirm creates and deletes by echo; docs and package metadata match the API | + +### Deferred until a read-only view needs them + +| Item | Trigger | Until then | +|---|---|---| +| `SnapshotTracker` extraction, `ResourceStore`, `readOnlyPolicy` removal and the read-only example | A list or viewer that should not load the merge engine, or a consumer asking for a read-only store | A list renders from `LiveResourceStore` with a read-only policy and carries a merge engine it does not use | +| `stream` and `editor` subpath entry points, the `sideEffects` audit and module-graph assertions | An unbundled consumer that measurably benefits from loading only the stream layer | Hosts import the root or the single-file bundle, which contain both layers | + +Each deferred item keeps the design and acceptance criteria in this document; build it as its own +change when its trigger appears. Use `feat!:` and short `BREAKING CHANGE:` notes for public removals. List connector/name/options changes, store input removals, response-redaction simplification, utility exports and completion rejections. Remove old names and signatures without overloads or forwarding aliases. Keep the -current -lockstep release process; generated versions and changelogs remain Release Please's responsibility. - -Migrate tests, conditional-save and Vue examples, browser example, READMEs, -adoption/state-model/save -and Vue guides, `url.ts` examples, and package metadata. Fix stale native EventSource descriptions -and -read-only policy/type comments. Lead with streaming and describe editing as optional; retain useful -merge keywords. A save success must not clear edits made while it was in flight. - -Update [spec §7](../../spec/v1.md#7-transport--authentication) to say the reference connector uses -fetch -for cookies and headers, and [consumer rule 9](../../spec/v1.md#10-conformance) to require closing -the -connection on terminal refusal rather than naming a JavaScript class. Keep gateway native SSE/cookie -support and a raw `EventSource` browser check. That check verifies framing, cookies, fresh snapshots -on native reconnect and terminal shutdown through test-owned handling; it is not another official -connector and does not claim browser-native retry timing satisfies the reference retry policy. -These are implementation-neutral wording changes, not a new wire version. Keep documented v1 event -fields and error codes, including `SLOW_CONSUMER`; their removal is separate protocol scope. +current lockstep release process; generated versions and changelogs remain Release Please's +responsibility. + +Change 1 migrated tests, the conditional-save, Vue and browser examples, READMEs, guides and +`url.ts` examples to the connector, fixed stale native EventSource descriptions, and added an +[upgrade guide](../migrating.md). Change 2 migrates store callers, package metadata and read-only +policy/type comments, and extends that guide. Lead with streaming and describe editing as optional; +retain useful merge keywords. A save success must not clear edits made while it was in flight. + +Change 1 updated [spec §7](../../spec/v1.md#7-transport--authentication) to say the reference +connector uses fetch for cookies and headers, and [consumer rule 9](../../spec/v1.md#10-conformance) +to require closing the connection on terminal refusal rather than naming a JavaScript class. Keep +gateway native SSE/cookie support and a raw `EventSource` browser check. That check verifies +framing, cookies, fresh snapshots on native reconnect and terminal shutdown through test-owned +handling; it is not another official connector and does not claim browser-native retry timing +satisfies the reference retry policy. These are implementation-neutral wording changes, not a new +wire version. Keep documented v1 event fields and error codes, including `SLOW_CONSUMER`; their +removal is separate protocol scope. ## Verification and completion @@ -445,34 +459,39 @@ Extend existing suites with observable outcomes: - Plain consumer, including a bound store method: chunked framing, ignored unknown events, gap discard and recovery, terminal refusal, retry hints/exhaustion, abort and repeated snapshot - cycles. - Verify state subscriptions replace lifecycle callbacks and gap context appears on retrying state. + cycles. Verify state subscriptions replace lifecycle callbacks and gap context appears on retrying + state. - Close from synced without a later live publication; close then throw without losing the original - exception; no retries after consumer failure; no pending reader, timer or listener after cleanup. + exception; no retries after an exception from the consumer, a subscriber or `onError`; no pending + reader, timer or listener after cleanup. - Matching, missing and mismatched protocol headers: mismatches apply no events and never retry. -- Both stores: replacement/field removal, detached data, redaction changes, repeated replay, - deletion while disconnected, same-name/new-UID replacement, partial/restarted snapshots, synced - without reset, and matching removal reports. Use authoritative-state conformance and convergence - fixtures for both stores and keep edit/conflict fixtures specific to the editor. +- Both stores, once `ResourceStore` lands: replacement/field removal, detached data, redaction + changes, repeated replay, deletion while disconnected, same-name/new-UID replacement, + partial/restarted snapshots, synced without reset, and matching removal reports. Use + authoritative-state conformance and convergence fixtures for both stores and keep edit/conflict + fixtures specific to the editor. - Editor: late GETs during and across completed recovery cycles, later typing during saves, refused - replacement UIDs, redaction-path omissions/unknown paths, conflict retention and keyed lists. - Reproduce the adopted-save ghost before removal; show that a guarded late response never counts - as membership and an empty completed snapshot prunes the UID. + replacement UIDs, redaction-path omissions/unknown paths, conflict retention and keyed lists. Keep + the adopted-save regression until `adoptSaved` is removed; show that a guarded late response never + counts as membership and an empty completed snapshot prunes the UID. - Host examples: no synthetic adoption/deletion, create confirmation in either echo/response order, pending-confirmation entries cannot be resubmitted, and a failed write leaves authoritative state unchanged. No reconciliation guard inserts a resource it did not capture. -- Packed exports, declarations and emitted graph: stream imports run with editor modules absent; - root and combined bundle expose the intended API. Real Go-wire and Chromium checks cover both - stores through the public connector and direct raw EventSource compatibility. +- Packed exports and declarations: root and combined bundle expose the intended API. Real Go-wire + and Chromium checks cover the editor through the public connector and direct raw EventSource + compatibility. When the deferred entry points land, the emitted graph shows stream imports run + with editor modules absent, and the checks cover both stores. For runtime PRs run `task fixtures-check`, `task test`, `task lint`, `task e2e-wire`, `task e2e-browser` and `task pack-client`, and inspect final-commit CI. Existing goldens must retain wire behavior. Real-cluster/capacity campaigns are not a browser-separation merge gate. Report the final tested commit, results, unrun host checks and limitations. -Completion means a read-only view loads no editor modules through the stream entry, a live editor -preserves draft/save guarantees, one public connector serves both, published artifacts work and -migration guidance matches the implementation. This document alone is not implementation evidence. +The planned work is complete when one public connector serves the editor, the editor's public input +is the event stream with no unguarded adoption, a live editor preserves draft and save guarantees, +published artifacts work and migration guidance matches the implementation. A read-only view that +loads no editor modules through a stream entry is the completion criterion of the deferred items. +This document alone is not implementation evidence. ## Relationship to the remaining roadmap @@ -482,6 +501,5 @@ upstream continuation. Align its recipes with this proposal's smaller API. Imple before expanding editor public surface; it does not block gateway fixes or real-API evidence. [Proposal 0010](0010-gateway-api-cleanup.md) owns Go cleanup and can proceed independently. [Proposal 0008](0008-shared-watch-hardening.md) remains completed; deferred authorization triggers -and -caches are not reopened. Let demonstrated stream and editing needs drive their respective +and caches are not reopened. Let demonstrated stream and editing needs drive their respective follow-ups. diff --git a/docs/proposals/0010-gateway-api-cleanup.md b/docs/proposals/0010-gateway-api-cleanup.md index e8f2edf..e2a365e 100644 --- a/docs/proposals/0010-gateway-api-cleanup.md +++ b/docs/proposals/0010-gateway-api-cleanup.md @@ -1,15 +1,15 @@ # Proposal 0010 Gateway API cleanup -**Status: proposed implementation plan.** No Go APIs are changed by this document. -Source review baseline: `a8281c5`; proposal review branch: `0cde13b`, 2026-10-04. +**Status: implemented** in [#54](https://github.com/ConfigButler/krm-stream/pull/54), as the two +changes below, pending review. Source review baseline: `a8281c5`; proposal review branch: `0cde13b`, +2026-10-04. Simplify gateway configuration and remove duplicate entry points and repository-only harness -exports. -Preserve authorization, projection, bounded HTTP delivery, shared-watch lifecycle and v1 wire -behavior. -Deliver independently from [browser separation](0009-stream-and-editor-separation.md). Follow the -[design rules](../../CONTRIBUTING.md#design-rules) and [release policy](../releasing.md): remove -superseded names before 1.0 and migrate callers together. +exports. Preserve authorization, projection, bounded HTTP delivery, shared-watch lifecycle and v1 +wire behavior. Deliver independently from [browser +separation](0009-stream-and-editor-separation.md). Follow the [design +rules](../../CONTRIBUTING.md#design-rules) and [release policy](../releasing.md): remove superseded +names before 1.0 and migrate callers together. The source audit establishes repository use, not every consumer's use. Field reports record krm-foyer adoption but do not establish which exports its current checkout imports. Inventory @@ -69,10 +69,10 @@ type Gateway struct { ``` `StreamConfig` contains `Authorizer`, `Clients`, `Projections`, `Ordering`, `Observer`, -`Diagnostics`, -`HeartbeatInterval`, `WriteTimeout`, `ReauthorizationInterval` and `ReauthorizationTimeout`. -Use Authorizer consistently, removing Gateway.Auth. Handler copies the single config value rather -than assigning every shared field individually. Preserve host-callback validation. +`Diagnostics`, `HeartbeatInterval`, `WriteTimeout`, `ReauthorizationInterval` and +`ReauthorizationTimeout`. Use Authorizer consistently, removing Gateway.Auth. Handler copies the +single config value rather than assigning every shared field individually. Preserve host-callback +validation. Go composite literals must explicitly name the embedded configuration: @@ -91,6 +91,10 @@ gateway.Options{ } ``` +The cost is one level of nesting in a host's literals. Assignments such as `options.Authorizer = …` +keep working through field promotion, and the [upgrade guide](../migrating.md) maps each old literal +to the new one. + Remove the Projection configuration field. Nil Projections retains the safe full policy through `StaticProjection(ProjectionFull)`. A host with one view supplies StaticProjection; a host selecting by principal/scope supplies ProjectionPolicy. Raw Secret disclosure continues to require explicit @@ -100,25 +104,26 @@ Preserve validation, including negative values and positive HTTP write timeout w The shared struct must not impose HTTP-only restrictions on transport-neutral sinks. Direct HTTP validation still occurs before response I/O or backend opening. Update comments and error messages to identify the shared Ordering setting without assuming Handler users configure a Gateway literal. +The strict-ordering refusal is the one message on the wire this touches: its golden transcript +changes in message prose only, which both suites compare loosely. ## Repository-only exports -Move [conformance.go](../../gateway/conformance.go) and -[scripted.go](../../gateway/scripted.go) into `gateway/internal/conformance` or similarly narrow -internal packages. Move Fixture, WatchOp, FixtureEvent, Corpus, loaders and ScriptedBackend together -as needed. The harness may import gateway domain types; production gateway must not import it. -Use explicit corpus paths from commands/tests. Remove the production relative-path convenience -loader; a test helper may supply the repository path. +Move [conformance.go](../../gateway/conformance.go) and [scripted.go](../../gateway/scripted.go) +into `gateway/internal/conformance` or similarly narrow internal packages. Move Fixture, WatchOp, +FixtureEvent, Corpus, loaders and ScriptedBackend together as needed. The harness may import gateway +domain types; production gateway must not import it. Use explicit corpus paths from commands/tests. +Remove the production relative-path convenience loader; a test helper may supply the repository +path. A `package gateway` test importing a harness that imports gateway creates a cycle. Move -fixture-driven -conformance and golden tests to `package gateway_test`, including fixture-dependent cases currently -in `stream_test.go`; adapt their recording/replay helpers. Leave private implementation tests in -`package gateway`. Do not convert the entire suite or export private helpers to avoid the cycle. -Inspect shared helper dependencies before editing. - -Keep replay and tests on the same fixture interpreter. Preserve use of the production SSE encoder -in golden generation, update flags, protocol.json generation, fixture paths and task names. Do not +fixture-driven conformance and golden tests to `package gateway_test`, including fixture-dependent +cases currently in `stream_test.go`; adapt their recording/replay helpers. Leave private +implementation tests in `package gateway`. Do not convert the entire suite or export private helpers +to avoid the cycle. Inspect shared helper dependencies before editing. + +Keep replay and tests on the same fixture interpreter. Preserve use of the production SSE encoder in +golden generation, update flags, protocol.json generation, fixture paths and task names. Do not replace executable wire evidence with manually encoded transcripts. Core gateway must no longer include repository corpus filesystem loading in its runtime module graph. @@ -139,8 +144,7 @@ canonical parser/TypeScript URL checks against the existing corpus. For generic sinks, clarify the existing requirement that the host bounds delivery and honors cancellation. Proposal 0008 repaired library-owned HTTP configuration; it did not promise to bound arbitrary host I/O. Keep generic bounded-sink tests separate from HTTP blocked-reader/deadline -tests. -Deleting the protocol seam is not needed to preserve the HTTP guarantee. +tests. Deleting the protocol seam is not needed to preserve the HTTP guarantee. Protocol removals require a separately reviewed compatibility/version decision. This cleanup removes duplicate configuration and forwarding paths while preserving the protocol and backend contract. @@ -156,9 +160,9 @@ Deliver two changes, each buildable and fully migrated: Update READMEs, adoption, auth and operations guides, root examples and kube examples. Historical proposals and dated field reports remain historical: update current-guidance links or add a short -migration note when useful, rather than rewriting the APIs they originally reviewed. -[Proposal 0006](0006-stream-and-save-implementation-plan.md) keeps its acceptance criteria and uses -the final method/configuration names in later work. +migration note when useful, rather than rewriting the APIs they originally reviewed. [Proposal +0006](0006-stream-and-save-implementation-plan.md) keeps its acceptance criteria and uses the final +method/configuration names in later work. Run `task fixtures-check`, `task test`, `task lint`, `task e2e-wire`, `task e2e-browser` and `task pack-client`. Run race checks in both Go modules for changed lifecycle paths. No new cluster @@ -169,8 +173,7 @@ Focused acceptance must establish: - Nil projection policy redacts Secrets; static policy refuses a different requested view; dynamic policy runs at opening, cycles and timed checks. - Handler/direct HTTP preserve fail-before-I/O validation, unsupported-writer refusal, - blocked-client - deadlines and healthy-peer isolation. Generic bounded sinks remain usable. + blocked-client deadlines and healthy-peer isolation. Generic bounded sinks remain usable. - Shared-backend zero/explicit options preserve cancellable openings, sharing, overflow recovery, backoff, last-subscriber cleanup and balanced observations. - Both config entry points use the same fields/names without per-field mapping. Current code and