From ec4f9b253a7d8c6bb6289a48c4078adffd3d98d6 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 5 Oct 2026 11:19:51 +0000 Subject: [PATCH 1/6] refactor(client): share the connection lifecycle and HTTP helpers between transports Move retries, the healthy-period reset, published state, cancellation and completion out of connectResourceStream into an internal runConnection that drives one transport function through TransportHooks. Move request options, refusal classification, the Kubernetes Status message reader and Retry-After parsing into an internal http module. The SSE connector's public API, classification and behavior are unchanged; its suite passes without edits beyond two import paths. Co-Authored-By: Claude Opus 5.5 --- packages/krm-stream/src/connection.ts | 225 ++---------------- packages/krm-stream/src/http.ts | 105 +++++++++ packages/krm-stream/src/lifecycle.ts | 243 ++++++++++++++++++++ packages/krm-stream/src/sse.ts | 125 +--------- packages/krm-stream/test/connection.test.ts | 4 +- packages/krm-stream/test/wire.test.ts | 2 +- 6 files changed, 381 insertions(+), 323 deletions(-) create mode 100644 packages/krm-stream/src/http.ts create mode 100644 packages/krm-stream/src/lifecycle.ts diff --git a/packages/krm-stream/src/connection.ts b/packages/krm-stream/src/connection.ts index c67db1b..5737eee 100644 --- a/packages/krm-stream/src/connection.ts +++ b/packages/krm-stream/src/connection.ts @@ -1,64 +1,24 @@ +import { + type ResourceEventConsumer, + type ResourceStreamHandle, + type ResourceStreamOptions, + runConnection, +} from "./lifecycle.ts"; import { streamOnce } from "./sse.ts"; -import type { ErrorCode, ResourceStateEvent } from "./types.ts"; -export type ConnectionStatus = "connecting" | "syncing" | "live" | "retrying" | "closed" | "terminal" | "exhausted"; - -export interface ConnectionState { - status: ConnectionStatus; - /** Retries since the last sustained healthy period. */ - retries: number; - retryInMs?: number; - /** Present only on a `retrying` state that a sequence gap caused: the `seq` this connection expected - * and the one it received. The event beyond the gap was discarded, and the retry asks for a fresh - * snapshot. */ - gap?: Readonly<{ expected: number; received: number }>; -} - -/** 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`. 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 { - /** Called for protocol and HTTP errors. A terminal error ends the stream without retry. - * `retryAfterMs` is the server's hint for a retryable error: an error event's `retryAfterMs`, or - * an HTTP `Retry-After`. */ - onError?: (code: ErrorCode, message: string, terminal: boolean, retryAfterMs?: number) => void; - /** Abort the stream from outside. */ - signal?: AbortSignal; - /** Defaults to same-origin; use include for a cross-origin cookie gateway. */ - credentials?: RequestCredentials; - /** Injectable for tests. Defaults to the global fetch. */ - fetch?: typeof globalThis.fetch; - headers?: Record; - /** Retry budget between sustained healthy periods. Defaults to 8. */ - maxRetries?: number; - /** Continuous live time required to reset retries and backoff. Defaults to 30 seconds. */ - healthyResetMs?: number; - /** Exponential backoff starts at 500ms, capped at 30s, with 50–100% jitter. */ - retryDelayMs?: number; - maxRetryDelayMs?: number; -} - -export interface ResourceStreamHandle { - /** The current state. Read it for the initial presentation; `subscribe` for every later one. */ - readonly state: Readonly; - 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, 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; -} - -/** Consume a resource stream over fetch, for both session cookies and bearer headers, delivering - * 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. +export type { + ConnectionState, + ConnectionStatus, + ResourceEventConsumer, + ResourceStreamHandle, + ResourceStreamOptions, +} from "./lifecycle.ts"; + +/** Consume a gateway resource stream over fetch, for both session cookies and bearer headers, + * delivering each state event to `consume`. Every reconnect requests a fresh snapshot; whatever the + * consumer holds survives it. Every HTTP 4xx except 408/429, 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. * * 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 @@ -75,150 +35,5 @@ export function connectResourceStream( consume: ResourceEventConsumer, opts: ResourceStreamOptions = {}, ): ResourceStreamHandle { - const maxRetries = opts.maxRetries ?? 8; - const healthyResetMs = opts.healthyResetMs ?? 30_000; - const delay = opts.retryDelayMs ?? 500; - const cap = opts.maxRetryDelayMs ?? 30_000; - if ( - !Number.isFinite(healthyResetMs) || - healthyResetMs <= 0 || - healthyResetMs > 2_147_483_647 || - !Number.isSafeInteger(maxRetries) || - maxRetries < 0 || - !Number.isFinite(delay) || - delay < 0 || - !Number.isFinite(cap) || - cap < 0 || - cap > 2_147_483_647 - ) { - throw new RangeError("krm-stream: invalid retry budget or delay"); - } - const controller = new AbortController(); - const subscribers = new Set<(state: Readonly) => void>(); - let state: Readonly = Object.freeze({ status: "connecting", retries: 0 }); - let terminal = false; - let hintMs: number | undefined; - let healthTimer: ReturnType | undefined; - const clearHealthTimer = () => { - 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) call(() => callback(state)); - }; - const close = () => { - clearHealthTimer(); - controller.abort(); - }; - 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 { - 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, - closed, - get state() { - return state; - }, - subscribe(callback) { - subscribers.add(callback); - return () => { - subscribers.delete(callback); - }; - }, - }; + return runConnection((hooks, signal) => streamOnce(url, hooks, signal, opts), consume, opts); } diff --git a/packages/krm-stream/src/http.ts b/packages/krm-stream/src/http.ts new file mode 100644 index 0000000..5667812 --- /dev/null +++ b/packages/krm-stream/src/http.ts @@ -0,0 +1,105 @@ +// HTTP helpers both transports share: request options, the classification of a refused response, +// the Kubernetes Status message a host proxy refuses with, and `Retry-After`. Framing lives in the +// transports; nothing here reads an event. + +import type { ErrorCode } from "./types.ts"; + +export interface RequestOptions { + /** Defaults to same-origin; use include for a cross-origin cookie gateway or proxy. */ + credentials?: RequestCredentials; + /** Injectable for tests. Defaults to the global fetch. */ + fetch?: typeof globalThis.fetch; + headers?: Record; +} + +/** A GET that a long-lived stream can rely on: cancellable, uncached and with the caller's + * credentials mode. A cached stream is a stream that never moves. */ +export function request(url: string, accept: string, signal: AbortSignal, opts: RequestOptions): Promise { + return (opts.fetch ?? globalThis.fetch)(url, { + signal, + headers: { Accept: accept, ...opts.headers }, + cache: "no-store", + credentials: opts.credentials ?? "same-origin", + }); +} + +/** How a refused HTTP status is reported: every 4xx except 408 and 429 is terminal, because asking + * again cannot change the answer and a client that does will hammer a scope it may never see. */ +export function refusal(status: number): { code: ErrorCode; terminal: boolean } { + const code: ErrorCode = + status === 401 + ? "UNAUTHENTICATED" + : status === 403 + ? "FORBIDDEN" + : status === 429 || status === 502 || status === 503 || status === 504 + ? "UPSTREAM_UNAVAILABLE" + : "INTERNAL"; + return { code, terminal: status >= 400 && status < 500 && status !== 408 && status !== 429 }; +} + +/** The largest refusal body read in search of a Kubernetes Status message. */ +const maxStatusBytes = 16 * 1024; +/** How long a refusal body may take to arrive. A known 401, 403 or 429 must not wait on a server + * that stalls the body explaining it. */ +const statusBudgetMs = 2_000; + +/** The `message` of a Kubernetes `Status` body, as a host proxying `/k8s` refuses with, or + * undefined. It reads at most maxStatusBytes for at most budgetMs, stops when `signal` aborts, and + * always releases the body. */ +export async function statusMessage( + res: Response, + signal: AbortSignal, + budgetMs = statusBudgetMs, +): Promise { + if (!res.body) return undefined; + if (signal.aborted || !/^application\/json\b/i.test(res.headers.get("Content-Type") ?? "")) { + await res.body.cancel().catch(() => {}); + return undefined; + } + const reader = res.body.getReader(); + let gaveUp = false; + // Cancelling the reader settles a pending read(), so a quiet body cannot hold this open. + const giveUp = () => { + gaveUp = true; + void reader.cancel().catch(() => {}); + }; + const timer = setTimeout(giveUp, budgetMs); + signal.addEventListener("abort", giveUp, { once: true }); + const chunks: Uint8Array[] = []; + let size = 0; + try { + for (;;) { + const { done, value } = await reader.read(); + if (gaveUp) return undefined; + if (done) break; + size += value.byteLength; + if (size > maxStatusBytes) return undefined; + chunks.push(value); + } + const bytes = new Uint8Array(size); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + const status: unknown = JSON.parse(new TextDecoder().decode(bytes)); + if (typeof status !== "object" || status === null) return undefined; + const { kind, message } = status as { kind?: unknown; message?: unknown }; + return kind === "Status" && typeof message === "string" && message !== "" ? message : undefined; + } catch { + return undefined; + } finally { + clearTimeout(timer); + signal.removeEventListener("abort", giveUp); + await reader.cancel().catch(() => {}); + } +} + +/** An HTTP `Retry-After` in milliseconds: delay-seconds or an HTTP-date. */ +export function retryAfter(header: string | null, now = Date.now()): number | undefined { + if (header === null) return undefined; + const value = header.trim(); + if (/^\d+$/.test(value)) return Number(value) * 1000; + const at = Date.parse(value); + return Number.isNaN(at) ? undefined : Math.max(0, at - now); +} diff --git a/packages/krm-stream/src/lifecycle.ts b/packages/krm-stream/src/lifecycle.ts new file mode 100644 index 0000000..b4562f7 --- /dev/null +++ b/packages/krm-stream/src/lifecycle.ts @@ -0,0 +1,243 @@ +// The connection lifecycle both connectors share: retries, the healthy-period reset, published state, +// cancellation and completion. It knows nothing about framing. A transport opens ONE connection, +// reports what happens on it through TransportHooks, and resolves when it ends; this file decides +// whether there is another. +// +// Internal on purpose. The two connectors in connection.ts and native.ts are the public surface; a +// general transport plugin API would be a contract nobody has asked for. + +import type { ErrorCode, ResourceStateEvent } from "./types.ts"; + +export type ConnectionStatus = "connecting" | "syncing" | "live" | "retrying" | "closed" | "terminal" | "exhausted"; + +export interface ConnectionState { + status: ConnectionStatus; + /** Retries since the last sustained healthy period. */ + retries: number; + retryInMs?: number; + /** Present only on a `retrying` state that a sequence gap caused: the `seq` this connection expected + * and the one it received. The event beyond the gap was discarded, and the retry asks for a fresh + * snapshot. */ + gap?: Readonly<{ expected: number; received: number }>; +} + +/** 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`. 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 { + /** Called for protocol and HTTP errors. A terminal error ends the stream without retry. + * `retryAfterMs` is the server's hint for a retryable error: an error event's `retryAfterMs`, a + * Kubernetes Status's `retryAfterSeconds`, or an HTTP `Retry-After`. */ + onError?: (code: ErrorCode, message: string, terminal: boolean, retryAfterMs?: number) => void; + /** Abort the stream from outside. */ + signal?: AbortSignal; + /** Defaults to same-origin; use include for a cross-origin cookie gateway or proxy. */ + credentials?: RequestCredentials; + /** Injectable for tests. Defaults to the global fetch. */ + fetch?: typeof globalThis.fetch; + headers?: Record; + /** Retry budget between sustained healthy periods. Defaults to 8. */ + maxRetries?: number; + /** Continuous live time required to reset retries and backoff. Defaults to 30 seconds. */ + healthyResetMs?: number; + /** Exponential backoff starts at 500ms, capped at 30s, with 50–100% jitter. */ + retryDelayMs?: number; + maxRetryDelayMs?: number; +} + +export interface ResourceStreamHandle { + /** The current state. Read it for the initial presentation; `subscribe` for every later one. */ + readonly state: Readonly; + 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, 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; +} + +/** What one connection reports to the lifecycle. A host observes the same moments through the + * handle's state, and the events themselves through its consumer. None of these throw: the lifecycle + * catches the host's exceptions itself and aborts the signal instead, so a transport checks + * `signal.aborted` after every call. */ +export interface TransportHooks { + /** Receives each state event synchronously, in stream order. */ + consume: (event: ResourceStateEvent) => void; + /** The server accepted the request. Its snapshot has not started yet. */ + opened(): void; + /** A snapshot is starting. Called before the consumer sees its `reset`. */ + reset(): void; + /** The consumer has applied `synced` and the connection is still open: the stream is live. */ + synced(): void; + /** A missing or duplicated event. The event beyond the gap was discarded; the connection ends. */ + gap(expected: number, received: number): void; + /** A protocol or HTTP error. A terminal one ends the connection and the stream. */ + error(code: ErrorCode, message: string, terminal: boolean, retryAfterMs?: number): void; +} + +/** Open one connection and report on it until it ends. It resolves in every case, after its readers + * and listeners are released, because whether to come back is the lifecycle's decision. */ +export type Transport = (hooks: TransportHooks, signal: AbortSignal) => Promise; + +/** Run `transport` until the stream ends, reconnecting within the bounded retry policy. + * + * 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. */ +export function runConnection( + transport: Transport, + consume: ResourceEventConsumer, + opts: ResourceStreamOptions, +): ResourceStreamHandle { + const maxRetries = opts.maxRetries ?? 8; + const healthyResetMs = opts.healthyResetMs ?? 30_000; + const delay = opts.retryDelayMs ?? 500; + const cap = opts.maxRetryDelayMs ?? 30_000; + if ( + !Number.isFinite(healthyResetMs) || + healthyResetMs <= 0 || + healthyResetMs > 2_147_483_647 || + !Number.isSafeInteger(maxRetries) || + maxRetries < 0 || + !Number.isFinite(delay) || + delay < 0 || + !Number.isFinite(cap) || + cap < 0 || + cap > 2_147_483_647 + ) { + throw new RangeError("krm-stream: invalid retry budget or delay"); + } + const controller = new AbortController(); + const subscribers = new Set<(state: Readonly) => void>(); + let state: Readonly = Object.freeze({ status: "connecting", retries: 0 }); + let terminal = false; + let hintMs: number | undefined; + let healthTimer: ReturnType | undefined; + const clearHealthTimer = () => { + 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) call(() => callback(state)); + }; + const close = () => { + clearHealthTimer(); + controller.abort(); + }; + 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 transport( + { + 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, + ); + 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 { + 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, + closed, + get state() { + return state; + }, + subscribe(callback) { + subscribers.add(callback); + return () => { + subscribers.delete(callback); + }; + }, + }; +} diff --git a/packages/krm-stream/src/sse.ts b/packages/krm-stream/src/sse.ts index bc9817d..e97a769 100644 --- a/packages/krm-stream/src/sse.ts +++ b/packages/krm-stream/src/sse.ts @@ -3,8 +3,8 @@ // // One connection at a time, over fetch: fetch sends the same-origin session cookie a v1 gateway must // accept (spec §7) and, unlike native EventSource, can also send `Authorization: Bearer`. It also -// works in Node, so it is what the conformance suite drives. connection.ts owns everything that spans -// connections — retries, lifecycle state and the public handle — and is the only caller. +// works in Node, so it is what the conformance suite drives. lifecycle.ts owns everything that spans +// connections — retries, lifecycle state and the public handle — and connection.ts is the only caller. // // The rule that is easy to get wrong: on a TERMINAL error, close the connection and do not come back. // A client that reconnects anyway will hammer a scope it can never be allowed to see — forever, from @@ -13,7 +13,9 @@ // Nothing here knows about a store. State events go to a consumer, and what it does with them is its // own business. -import type { ErrorCode, ResourceStateEvent, StreamEvent } from "./types.ts"; +import { type RequestOptions, refusal, request, retryAfter, statusMessage } from "./http.ts"; +import type { TransportHooks } from "./lifecycle.ts"; +import type { ResourceStateEvent, StreamEvent } from "./types.ts"; import { PROTOCOL_VERSION } from "./version.ts"; /** Incremental SSE parser. Bytes arrive in whatever chunks the network feels like — a frame can be @@ -118,32 +120,6 @@ 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. 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; - /** The gateway accepted the stream. Its snapshot has not started yet. */ - opened(): void; - /** A `reset` arrived. Called before the consumer sees it. */ - reset(): void; - /** The consumer has applied `synced` and the connection is still open. */ - synced(): void; - /** A missing or duplicated event. The event beyond the gap was discarded; the connection ends. */ - gap(expected: number, received: number): void; - /** A protocol or HTTP error. A terminal one ends the connection. */ - error(code: ErrorCode, message: string, terminal: boolean, retryAfterMs?: number): void; -} - -export interface ConnectionOptions { - /** Defaults to same-origin; use include for a cross-origin cookie gateway. */ - credentials?: RequestCredentials; - /** Injectable for tests. Defaults to the global fetch. */ - fetch?: typeof globalThis.fetch; - headers?: Record; -} - /** Open one fetch connection and deliver its state events until it ends: EOF, a network failure, a * refusal, a terminal error, a sequence gap or `signal`. * @@ -151,9 +127,9 @@ export interface ConnectionOptions { * back is the caller's decision. */ export async function streamOnce( url: string, - hooks: ConnectionHooks, + hooks: TransportHooks, signal: AbortSignal, - opts: ConnectionOptions = {}, + opts: RequestOptions = {}, ): Promise { const controller = new AbortController(); const abort = () => controller.abort(); @@ -191,23 +167,9 @@ export async function streamOnce( try { if (controller.signal.aborted) return; - const res = await (opts.fetch ?? globalThis.fetch)(url, { - signal: controller.signal, - headers: { Accept: "text/event-stream", ...opts.headers }, - // The stream IS the response body; a cached one is a stream that never moves. - cache: "no-store", - credentials: opts.credentials ?? "same-origin", - }); + const res = await request(url, "text/event-stream", controller.signal, opts); if (!res.ok || !res.body) { - const code: ErrorCode = - res.status === 401 - ? "UNAUTHENTICATED" - : res.status === 403 - ? "FORBIDDEN" - : res.status === 429 || res.status === 502 || res.status === 503 || res.status === 504 - ? "UPSTREAM_UNAVAILABLE" - : "INTERNAL"; - const terminal = res.status >= 400 && res.status < 500 && res.status !== 408 && res.status !== 429; + const { code, terminal } = refusal(res.status); const message = (await statusMessage(res, controller.signal)) ?? `stream: HTTP ${res.status}`; if (controller.signal.aborted) return; // closed while reading the refusal: report nothing hooks.error(code, message, terminal, retryAfter(res.headers.get("Retry-After"))); @@ -258,75 +220,8 @@ export async function streamOnce( } } catch { // The network failed. This connection is over, and the caller decides about the next. Host - // callbacks never land here: connectResourceStream catches their exceptions itself. + // callbacks never land here: the lifecycle catches their exceptions itself. } finally { signal.removeEventListener("abort", abort); } } - -/** The largest refusal body read in search of a Kubernetes Status message. */ -const maxStatusBytes = 16 * 1024; -/** How long a refusal body may take to arrive. A known 401, 403 or 429 must not wait on a server - * that stalls the body explaining it. */ -const statusBudgetMs = 2_000; - -/** The `message` of a Kubernetes `Status` body, as a host proxying `/k8s` refuses with, or - * undefined. It reads at most maxStatusBytes for at most budgetMs, stops when `signal` aborts, and - * always releases the body. */ -export async function statusMessage( - res: Response, - signal: AbortSignal, - budgetMs = statusBudgetMs, -): Promise { - if (!res.body) return undefined; - if (signal.aborted || !/^application\/json\b/i.test(res.headers.get("Content-Type") ?? "")) { - await res.body.cancel().catch(() => {}); - return undefined; - } - const reader = res.body.getReader(); - let gaveUp = false; - // Cancelling the reader settles a pending read(), so a quiet body cannot hold this open. - const giveUp = () => { - gaveUp = true; - void reader.cancel().catch(() => {}); - }; - const timer = setTimeout(giveUp, budgetMs); - signal.addEventListener("abort", giveUp, { once: true }); - const chunks: Uint8Array[] = []; - let size = 0; - try { - for (;;) { - const { done, value } = await reader.read(); - if (gaveUp) return undefined; - if (done) break; - size += value.byteLength; - if (size > maxStatusBytes) return undefined; - chunks.push(value); - } - const bytes = new Uint8Array(size); - let offset = 0; - for (const chunk of chunks) { - bytes.set(chunk, offset); - offset += chunk.byteLength; - } - const status: unknown = JSON.parse(new TextDecoder().decode(bytes)); - if (typeof status !== "object" || status === null) return undefined; - const { kind, message } = status as { kind?: unknown; message?: unknown }; - return kind === "Status" && typeof message === "string" && message !== "" ? message : undefined; - } catch { - return undefined; - } finally { - clearTimeout(timer); - signal.removeEventListener("abort", giveUp); - await reader.cancel().catch(() => {}); - } -} - -/** An HTTP `Retry-After` in milliseconds: delay-seconds or an HTTP-date. */ -export function retryAfter(header: string | null, now = Date.now()): number | undefined { - if (header === null) return undefined; - const value = header.trim(); - if (/^\d+$/.test(value)) return Number(value) * 1000; - const at = Date.parse(value); - return Number.isNaN(at) ? undefined : Math.max(0, at - now); -} diff --git a/packages/krm-stream/test/connection.test.ts b/packages/krm-stream/test/connection.test.ts index 29e34da..a6436cd 100644 --- a/packages/krm-stream/test/connection.test.ts +++ b/packages/krm-stream/test/connection.test.ts @@ -823,7 +823,7 @@ for (const [name, prefix] of [ ["partial", '{"kind":"Status","mess'], ] as const) test(`a ${name} refusal body falls back to the bare status within its budget`, async () => { - const { statusMessage } = await import("../src/sse.ts"); + const { statusMessage } = await import("../src/http.ts"); const { response, body } = stalledRefusal(prefix); const started = Date.now(); assert.equal(await statusMessage(response, new AbortController().signal, 20), undefined); @@ -832,7 +832,7 @@ for (const [name, prefix] of [ }); test("aborting during a refusal-body read stops it at once", async () => { - const { statusMessage } = await import("../src/sse.ts"); + const { statusMessage } = await import("../src/http.ts"); const { response, body } = stalledRefusal('{"kind":'); const abort = new AbortController(); const read = statusMessage(response, abort.signal, 60_000); diff --git a/packages/krm-stream/test/wire.test.ts b/packages/krm-stream/test/wire.test.ts index 568c22d..f837be2 100644 --- a/packages/krm-stream/test/wire.test.ts +++ b/packages/krm-stream/test/wire.test.ts @@ -163,7 +163,7 @@ function decodeInChunks(bytes: string, size: number): StreamEvent[] { } test("Retry-After accepts delay-seconds and an HTTP-date", async () => { - const { retryAfter } = await import("../src/sse.ts"); + const { retryAfter } = await import("../src/http.ts"); const now = Date.parse("2026-10-02T12:00:00Z"); assert.equal(retryAfter("3", now), 3000); assert.equal(retryAfter("Fri, 02 Oct 2026 12:00:05 GMT", now), 5000); From f94c918efabc5488fdb1c4021353980354e128d9 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 5 Oct 2026 11:19:51 +0000 Subject: [PATCH 2/6] feat(client): watch native Kubernetes collections through a host proxy Add connectNativeWatch and nativeCollectionURL. Each connection performs an ordinary unpaginated LIST, delivers reset and the complete membership, opens a WATCH from the collection resourceVersion with the same selectors, and only then delivers synced and publishes live. Native watch JSON is decoded incrementally across byte, line and UTF-8 boundaries; bookmarks change nothing. Every reconnect re-lists, so the store keeps previous state until a completed snapshot prunes it. HTTP or in-stream 410, 408, 429, 5xx, network failures, EOF and malformed or truncated input use the shared bounded retry policy and hints; 401, 403 and other 4xx are terminal, as is a LIST carrying a continuation token, which is refused before reset. Read-only: no native editing, resume, streaming lists or pagination in this slice. Co-Authored-By: Claude Opus 5.5 --- packages/krm-stream/src/index.ts | 12 +- packages/krm-stream/src/native.ts | 398 ++++++++++ packages/krm-stream/src/url.ts | 41 ++ packages/krm-stream/test/native.test.ts | 927 ++++++++++++++++++++++++ 4 files changed, 1374 insertions(+), 4 deletions(-) create mode 100644 packages/krm-stream/src/native.ts create mode 100644 packages/krm-stream/test/native.test.ts diff --git a/packages/krm-stream/src/index.ts b/packages/krm-stream/src/index.ts index 9101f06..da4e9f9 100644 --- a/packages/krm-stream/src/index.ts +++ b/packages/krm-stream/src/index.ts @@ -2,11 +2,14 @@ // // The public surface is small on purpose: // -// connectResourceStream a conforming SSE consumer: delivers each resource state event to a -// callback, and recovers with bounded retries and a fresh snapshot. +// connectResourceStream a conforming gateway SSE consumer: delivers each resource state event to +// a callback, and recovers with bounded retries and a fresh snapshot. +// connectNativeWatch the same lifecycle and events over a native Kubernetes LIST and WATCH, +// through a proxy the host already runs. // LiveResourceStore holds server truth + your draft; three-way merges every watch event; // derives dirtiness; tracks conflicts; builds the merge patch. // resourceStreamURL builds the stream URL from a scope — the encoding the gateway parses back. +// nativeCollectionURL builds a native collection URL behind a host proxy. // // The store is built test-first against conformance/ — the same fixtures the Go gateway runs. See // ../../../docs/client-state-model.md for the algorithm, ../../../spec/v1.md for the wire, and @@ -25,6 +28,7 @@ export type { export { connectResourceStream } from "./connection.ts"; // Useful to a host that renders paths, and to anyone writing a policy: identity is a segment ARRAY. export { clone, deepEqual } from "./deep.ts"; +export { connectNativeWatch } from "./native.ts"; export { get, has, isPrefix, parsePointer, pathKey } from "./path.ts"; export { DEFAULT_EDITABLE_REGIONS, defaultPolicy, readOnlyPolicy, regionPolicy } from "./policy.ts"; export type { KubernetesStructuralSchema } from "./schema.ts"; @@ -47,6 +51,6 @@ export type { Scope, StreamEvent, } from "./types.ts"; -export type { ScopeQuery } from "./url.ts"; -export { resourceStreamURL } from "./url.ts"; +export type { NativeScope, ScopeQuery } from "./url.ts"; +export { nativeCollectionURL, resourceStreamURL } from "./url.ts"; export { PROTOCOL_VERSION, VERSION } from "./version.ts"; diff --git a/packages/krm-stream/src/native.ts b/packages/krm-stream/src/native.ts new file mode 100644 index 0000000..7e16326 --- /dev/null +++ b/packages/krm-stream/src/native.ts @@ -0,0 +1,398 @@ +// The native transport: an ordinary Kubernetes LIST, then a WATCH from the collection's +// resourceVersion, through a proxy the host already runs. No gateway, no SSE, no projection: the +// objects are exactly what the proxy returns, and the host owns credentials, routing and +// authorization. +// +// Every connection starts with a fresh, complete LIST, so a reconnect needs no checkpoint: the +// snapshot repairs whatever the previous watch missed — deletes, selector exits, a UID replaced +// under the same name — and the store prunes only once it completes. Resumable watches, streaming +// lists and pagination are later work with their own recovery rules; this file deliberately +// implements none of them. +// +// lifecycle.ts owns retries, state and cancellation, shared with the gateway connector. This file +// owns framing and its classification, and imports nothing from the SSE side: there is no `seq`, no +// protocol header and no redaction here. + +import { refusal, request, retryAfter, statusMessage } from "./http.ts"; +import { + type ResourceEventConsumer, + type ResourceStreamHandle, + type ResourceStreamOptions, + runConnection, + type TransportHooks, +} from "./lifecycle.ts"; +import type { ErrorCode, Identity, KRMObject } from "./types.ts"; + +/** Query parameters the connector sets itself, refused in a caller's collection URL. */ +const reserved = ["watch", "resourceVersion", "resourceVersionMatch", "limit", "continue", "sendInitialEvents"]; + +/** Watch a native Kubernetes collection through a host proxy, delivering each state event to + * `consume`. `collectionURL` is a collection path the proxy serves, such as one + * nativeCollectionURL builds; its selectors apply to both the LIST and the WATCH. + * + * Each connection LISTs the complete collection and delivers `reset`, an `added` per member, and — + * once the WATCH from the collection's resourceVersion is accepted — `synced`, after which the state + * is `live` and watch events follow as `added`, `modified` and `deleted`. Bookmarks change nothing. + * Every reconnect LISTs again, so a store keeps its previous state until the next snapshot completes + * and then prunes what it no longer contains. + * + * HTTP or in-stream 410 (history expired), 408, 429, 5xx, network failures, EOF and malformed or + * truncated frames consume the bounded retry budget, honoring `Retry-After` and a Status's + * `retryAfterSeconds`. HTTP or in-stream 401, 403 and every other 4xx are terminal, as is a LIST + * response that is only one page of the collection. Host exceptions, `close()` and `signal` behave + * exactly as for connectResourceStream. + * + * Objects arrive as the proxy returns them, Secret values and machinery fields included: native + * access provides no projection, redaction, suppression or watch sharing. A source that refused a + * projected stream must never be replaced with this one, and separate sources, scopes and identities + * need separate stores. */ +export function connectNativeWatch( + collectionURL: string, + consume: ResourceEventConsumer, + opts: ResourceStreamOptions = {}, +): ResourceStreamHandle { + const url = collectionURL.split("#")[0]!; + const query = new URLSearchParams(url.includes("?") ? url.slice(url.indexOf("?") + 1) : ""); + for (const name of reserved) { + if (query.has(name)) throw new Error(`krm-stream: the collection URL must not set ${name}; the connector does`); + } + return runConnection((hooks, signal) => watchOnce(url, hooks, signal, opts), consume, opts); +} + +/** The apiVersion and kind a typed collection's items have: `ConfigMapList` lists `ConfigMap`s. */ +interface ItemType { + apiVersion: string; + kind: string; +} + +/** A frame or a response this connector cannot use. Reported, never skipped: the connection ends + * and the next one re-lists. */ +class Malformed extends Error {} + +/** One connection: LIST, then WATCH until it ends. Resolves in every case, after the readers and + * listeners are released. */ +async function watchOnce( + url: string, + hooks: TransportHooks, + signal: AbortSignal, + opts: ResourceStreamOptions, +): Promise { + const controller = new AbortController(); + const abort = () => controller.abort(); + signal.addEventListener("abort", abort, { once: true }); + if (signal.aborted) abort(); + const aborted = () => controller.signal.aborted; + // Called detached, so a consumer never sees this hooks object as `this`. + const { consume } = hooks; + + /** Report a refused response; false when the connection must end. */ + const refused = async (phase: string, res: Response) => { + const { code, terminal } = classify(res.status); + const message = (await statusMessage(res, controller.signal)) ?? `native ${phase}: HTTP ${res.status}`; + if (aborted()) return; // closed while reading the refusal: report nothing + hooks.error(code, message, terminal, retryAfter(res.headers.get("Retry-After"))); + }; + const malformed = (phase: string, error: Malformed) => { + if (!aborted()) hooks.error("INTERNAL", `native ${phase}: ${error.message}`, false); + }; + + try { + if (aborted()) return; + const listed = await request(url, "application/json", controller.signal, opts); + if (!listed.ok || !listed.body) return await refused("list", listed); + if (aborted()) return void (await listed.body.cancel().catch(() => {})); + hooks.opened(); + + let collection: Collection; + try { + const text = await readText(listed.body, controller.signal); + if (text === undefined) return; + collection = readCollection(text); + } catch (error) { + if (error instanceof Unpaginated) { + if (!aborted()) hooks.error("INTERNAL", error.message, true); + return; + } + if (error instanceof Malformed) return malformed("list", error); + throw error; // the network failed while the body arrived + } + + // Membership is established only by a COMPLETE collection, so nothing above this line delivers. + hooks.reset(); + if (aborted()) return; + consume({ type: "reset" }); + for (const object of collection.items) { + if (aborted()) return; + consume({ type: "added", object }); + } + if (aborted()) return; + + const watched = await request( + `${url}${url.includes("?") ? "&" : "?"}watch=1&resourceVersion=${encodeURIComponent(collection.resourceVersion)}`, + "application/json", + controller.signal, + opts, + ); + if (!watched.ok || !watched.body) return await refused("watch", watched); + if (aborted()) return void (await watched.body.cancel().catch(() => {})); + // The snapshot is applied and the watch is accepted: only now is the state complete and live. + consume({ type: "synced" }); + // A consumer that closed the stream — or threw — while applying `synced` gets no later `live`. + if (aborted()) return void (await watched.body.cancel().catch(() => {})); + hooks.synced(); + + const reader = watched.body.getReader(); + const cancelReader = () => { + void reader.cancel().catch(() => {}); + }; + controller.signal.addEventListener("abort", cancelReader, { once: true }); + if (aborted()) cancelReader(); + const decoder = new WatchDecoder(); + try { + for (;;) { + const { done, value } = await reader.read(); + if (aborted()) return; + let lines: string[]; + try { + if (done) return decoder.end(); + lines = decoder.push(value); + } catch (error) { + if (!(error instanceof Malformed)) throw error; + return malformed("watch", error); + } + for (const line of lines) { + // One chunk can hold many frames. Nothing after a close is delivered. + if (aborted()) return; + let more: boolean; + try { + more = deliver(line, collection.type, hooks); + } catch (error) { + if (!(error instanceof Malformed)) throw error; + return malformed("watch", error); + } + if (!more) return; + } + } + } finally { + controller.signal.removeEventListener("abort", cancelReader); + await reader.cancel().catch(() => {}); + } + } catch { + // The network failed. This connection is over, and the lifecycle decides about the next. Host + // callbacks never land here: the lifecycle catches their exceptions itself. + } finally { + controller.abort(); + signal.removeEventListener("abort", abort); + } +} + +/** Apply one watch frame; false when the connection must end. Throws Malformed. */ +function deliver(line: string, type: ItemType | undefined, hooks: TransportHooks): boolean { + let frame: unknown; + try { + frame = JSON.parse(line); + } catch { + throw new Malformed("a watch frame is not JSON"); + } + if (!isRecord(frame) || typeof frame.type !== "string" || !isRecord(frame.object)) { + throw new Malformed("a watch frame has no type or object"); + } + const object = frame.object; + switch (frame.type) { + case "ADDED": + hooks.consume({ type: "added", object: resource(object, type, "watch") }); + return true; + case "MODIFIED": + hooks.consume({ type: "modified", object: resource(object, type, "watch") }); + return true; + case "DELETED": + hooks.consume({ type: "deleted", identity: identity(resource(object, type, "watch")) }); + return true; + case "BOOKMARK": + // A collection checkpoint, not an object version: it changes neither membership nor any + // object. Every reconnect re-lists, so there is nothing to resume from it either. + return true; + case "ERROR": { + const { code, terminal, message, retryAfterMs } = statusError(object); + hooks.error(code, message, terminal, retryAfterMs); + // The API server ends the watch after an error; either way, the next connection re-lists. + return false; + } + default: + throw new Malformed(`unknown watch event type ${JSON.stringify(frame.type)}`); + } +} + +/** How a native HTTP status or Status code is reported. Unlike the gateway's HTTP 410, a native 410 + * means the watch history expired, which a fresh LIST recovers. */ +function classify(status: number): { code: ErrorCode; terminal: boolean } { + return status === 410 ? { code: "RESYNC_REQUIRED", terminal: false } : refusal(status); +} + +/** A watch ERROR event's Kubernetes Status, classified like the HTTP status it carries. */ +function statusError(status: Record) { + const reason = typeof status.reason === "string" ? status.reason : undefined; + const code = + typeof status.code === "number" ? status.code : reason === "Expired" || reason === "Gone" ? 410 : undefined; + const { code: errorCode, terminal } = + code === undefined ? { code: "INTERNAL" as const, terminal: false } : classify(code); + const text = typeof status.message === "string" ? status.message : ""; + const details = status.details; + const seconds = isRecord(details) ? details.retryAfterSeconds : undefined; + return { + code: errorCode, + terminal, + message: text !== "" ? text : `native watch: error ${code ?? reason ?? "without a status"}`, + retryAfterMs: typeof seconds === "number" && seconds >= 0 ? seconds * 1000 : undefined, + }; +} + +interface Collection { + items: KRMObject[]; + resourceVersion: string; + type: ItemType | undefined; +} + +/** A LIST response that is one page of a larger collection. Pruning from it would delete every + * object on the other pages, so it is refused rather than retried. */ +class Unpaginated extends Error {} + +/** Parse and validate a complete collection. Throws Unpaginated or Malformed. */ +function readCollection(text: string): Collection { + let body: unknown; + try { + body = JSON.parse(text); + } catch { + throw new Malformed("the collection is not JSON"); + } + if (!isRecord(body)) throw new Malformed("the collection is not an object"); + const metadata = isRecord(body.metadata) ? body.metadata : {}; + if (typeof metadata.continue === "string" && metadata.continue !== "") { + throw new Unpaginated( + "native list: the response is one page of a paginated collection, which this connector does not support", + ); + } + const resourceVersion = metadata.resourceVersion; + if (typeof resourceVersion !== "string" || resourceVersion === "") { + throw new Malformed("the collection has no metadata.resourceVersion"); + } + const raw = body.items; + if (raw !== null && !Array.isArray(raw)) throw new Malformed("the collection has no items array"); + const type = itemType(body); + const uids = new Set(); + const items = (raw ?? []).map((item: unknown) => { + const object = resource(item, type, "list"); + if (uids.has(object.metadata.uid)) throw new Malformed(`the collection lists UID ${object.metadata.uid} twice`); + uids.add(object.metadata.uid); + return object; + }); + return { items, resourceVersion, type }; +} + +/** The item type of a concrete typed collection, or undefined. Only a `List` names its items' + * kind; a generic `List`, a `Table` or a `meta.k8s.io` collection does not, and a kind is never + * guessed from a plural resource name. */ +function itemType(collection: Record): ItemType | undefined { + const { apiVersion, kind } = collection; + if (typeof apiVersion !== "string" || apiVersion === "" || apiVersion.startsWith("meta.k8s.io/")) return undefined; + if (typeof kind !== "string" || !/^[A-Za-z0-9]+List$/.test(kind)) return undefined; + return { apiVersion, kind: kind.slice(0, -"List".length) }; +} + +/** A complete object with the identity a store keys on. Type metadata already present is kept; + * missing type metadata comes from the collection's item type. Throws Malformed. */ +function resource(value: unknown, type: ItemType | undefined, phase: "list" | "watch"): KRMObject { + const where = phase === "list" ? "a collection item" : "a watch object"; + if (!isRecord(value) || !isRecord(value.metadata)) throw new Malformed(`${where} has no metadata`); + const { uid, name, namespace } = value.metadata; + if (typeof uid !== "string" || uid === "") throw new Malformed(`${where} has no metadata.uid`); + if (typeof name !== "string" || name === "") throw new Malformed(`${where} ${uid} has no metadata.name`); + if (namespace !== undefined && typeof namespace !== "string") { + throw new Malformed(`${where} ${uid} has an invalid metadata.namespace`); + } + const present = (field: unknown) => typeof field === "string" && field !== ""; + if (present(value.apiVersion) && present(value.kind)) return value as KRMObject; + const apiVersion = present(value.apiVersion) ? value.apiVersion : type?.apiVersion; + const kind = present(value.kind) ? value.kind : type?.kind; + if (apiVersion === undefined || kind === undefined) { + throw new Malformed(`${where} ${uid} has no apiVersion or kind, and the collection does not name its item type`); + } + return { ...value, apiVersion, kind } as KRMObject; +} + +function identity(object: KRMObject): Identity { + const { uid, name, namespace } = object.metadata; + return { + uid, + apiVersion: object.apiVersion, + kind: object.kind, + ...(namespace === undefined ? {} : { namespace }), + name, + }; +} + +/** Incremental decoder for native watch JSON: one event per line. Bytes arrive in arbitrary chunks, + * so a frame — or a multi-byte character — can be split anywhere; this buffers until a line is + * complete. Throws Malformed on invalid UTF-8 and when the stream ends mid-frame. */ +export class WatchDecoder { + #utf8 = new TextDecoder("utf-8", { fatal: true }); + #buffer = ""; + + /** Feed a chunk; get back the complete, nonblank lines it finished. */ + push(chunk: Uint8Array): string[] { + this.#buffer += decode(this.#utf8, chunk); + const lines: string[] = []; + for (;;) { + const end = this.#buffer.indexOf("\n"); + if (end === -1) break; + const line = this.#buffer.slice(0, end).trim(); + this.#buffer = this.#buffer.slice(end + 1); + if (line !== "") lines.push(line); + } + return lines; + } + + /** The stream ended. Anything still buffered is a truncated frame. */ + end(): void { + this.#buffer += decode(this.#utf8); + if (this.#buffer.trim() !== "") throw new Malformed("the watch ended inside a frame"); + } +} + +/** A response body as text, or undefined once `signal` aborts. The body is always released. Throws + * Malformed on invalid UTF-8. */ +async function readText(body: ReadableStream, signal: AbortSignal): Promise { + const reader = body.getReader(); + const cancel = () => { + void reader.cancel().catch(() => {}); + }; + signal.addEventListener("abort", cancel, { once: true }); + if (signal.aborted) cancel(); + const utf8 = new TextDecoder("utf-8", { fatal: true }); + let text = ""; + try { + for (;;) { + const { done, value } = await reader.read(); + if (signal.aborted) return undefined; + if (done) return text + decode(utf8); + text += decode(utf8, value); + } + } finally { + signal.removeEventListener("abort", cancel); + await reader.cancel().catch(() => {}); + } +} + +/** Decode a chunk, or flush the decoder when there is none. Invalid or incomplete UTF-8 is malformed + * input, never a replacement character in a resource. */ +function decode(utf8: TextDecoder, chunk?: Uint8Array): string { + try { + return chunk === undefined ? utf8.decode() : utf8.decode(chunk, { stream: true }); + } catch { + throw new Malformed("the response is not valid UTF-8"); + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/packages/krm-stream/src/url.ts b/packages/krm-stream/src/url.ts index 5f174e5..8765aff 100644 --- a/packages/krm-stream/src/url.ts +++ b/packages/krm-stream/src/url.ts @@ -50,3 +50,44 @@ export function resourceStreamURL(base: string, scope: ScopeQuery): string { return `${base}${base.includes("?") ? "&" : "?"}${q.toString()}`; } + +/** A native collection: what nativeCollectionURL needs to address it through a host proxy. An + * empty or absent `group` is the core API; an empty or absent `namespace` is cluster-wide (or every + * namespace, for a namespaced resource the host allows that for). */ +export type NativeScope = Omit; + +/** + * Build the URL of a native Kubernetes collection behind a host proxy, for connectNativeWatch. + * + * ```ts + * connectNativeWatch(nativeCollectionURL("/k8s", { + * group: "apps", version: "v1", resource: "deployments", namespace: "app", labelSelector: "tier=web", + * }), (event) => applyStreamEvent(store, event)); + * // → /k8s/apis/apps/v1/namespaces/app/deployments?labelSelector=tier%3Dweb + * ``` + * + * `proxyBase` is wherever the host mounts the API server's paths: `/k8s` for a proxy at + * `/k8s/api/v1/...`, or an empty string for one at the origin's root. A single object is watched by + * name with `fieldSelector=metadata.name=`, so it can be absent, deleted and recreated like + * any other collection member. The host owns credentials, routing and which collections it allows. + */ +export function nativeCollectionURL(proxyBase: string, scope: NativeScope): string { + const segment = (what: string, value: string | undefined) => { + if (!value || value === "." || value === ".." || value.includes("/")) { + throw new Error(`krm-stream: invalid ${what} ${JSON.stringify(value ?? "")}`); + } + return encodeURIComponent(value); + }; + const path = [proxyBase.replace(/\/+$/, "")]; + if (scope.group) path.push("apis", segment("group", scope.group)); + else path.push("api"); + path.push(segment("version", scope.version)); + if (scope.namespace) path.push("namespaces", segment("namespace", scope.namespace)); + path.push(segment("resource", scope.resource)); + + const q = new URLSearchParams(); + if (scope.labelSelector) q.append("labelSelector", scope.labelSelector); + if (scope.name) q.append("fieldSelector", `metadata.name=${scope.name}`); + const query = q.toString(); + return path.join("/") + (query ? `?${query}` : ""); +} diff --git a/packages/krm-stream/test/native.test.ts b/packages/krm-stream/test/native.test.ts new file mode 100644 index 0000000..d30b513 --- /dev/null +++ b/packages/krm-stream/test/native.test.ts @@ -0,0 +1,927 @@ +// The native connector against a scripted fake fetch: every LIST and WATCH request is answered by +// the next step of a script, so each test decides exactly what the "API server" says and in which +// chunks. The lifecycle itself is shared with the gateway connector and covered in +// connection.test.ts; these tests cover what native transport adds and how it maps onto that +// lifecycle. + +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { + applyStreamEvent, + type ConnectionState, + type ConnectionStatus, + connectNativeWatch, + LiveResourceStore, + nativeCollectionURL, + type ResourceStateEvent, + type ResourceStreamHandle, + readOnlyPolicy, +} from "../src/index.ts"; + +const encoder = new TextEncoder(); +const ignore = () => {}; + +const cm = (uid: string, name: string, rv: string, value = "v") => ({ + apiVersion: "v1", + kind: "ConfigMap", + metadata: { uid, name, namespace: "app", resourceVersion: rv }, + data: { value }, +}); +const json = (body: unknown, status = 200, headers: Record = {}) => + new Response(JSON.stringify(body), { status, headers: { "Content-Type": "application/json", ...headers } }); +const collection = ( + items: unknown[], + rv = "10", + extra: { kind?: string; apiVersion?: string; continue?: string } = {}, +) => + json({ + kind: extra.kind ?? "ConfigMapList", + apiVersion: extra.apiVersion ?? "v1", + metadata: { resourceVersion: rv, ...(extra.continue === undefined ? {} : { continue: extra.continue }) }, + items, + }); +const status = (code: number, message: string, extra: Record = {}) => ({ + kind: "Status", + apiVersion: "v1", + status: "Failure", + message, + code, + ...extra, +}); +const frames = (...events: unknown[]) => events.map((e) => `${JSON.stringify(e)}\n`).join(""); + +/** A watch body. `chunk` splits it every that many BYTES, so frames and characters split anywhere. + * `open` keeps it open after the text; `push` and `end` drive it later. */ +function body(text: string, opts: { chunk?: number; open?: boolean } = {}) { + const state = { cancelled: false }; + let controller!: ReadableStreamDefaultController; + const bytes = encoder.encode(text); + const size = opts.chunk ?? Math.max(bytes.length, 1); + const response = new Response( + new ReadableStream({ + start(c) { + controller = c; + for (let i = 0; i < bytes.length; i += size) c.enqueue(bytes.slice(i, i + size)); + if (!opts.open) c.close(); + }, + cancel() { + state.cancelled = true; + }, + }), + ); + return { + response, + state, + push: (more: string | Uint8Array) => controller.enqueue(typeof more === "string" ? encoder.encode(more) : more), + end: () => controller.close(), + }; +} + +type Step = (url: string, signal: AbortSignal) => Response | Promise; +/** Answer the n-th request with the n-th step; a request beyond the script fails like the network. */ +function scripted(...steps: Step[]) { + const urls: string[] = []; + const fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + urls.push(String(input)); + const step = steps[urls.length - 1]; + if (!step) throw new TypeError("offline"); + return step(String(input), init!.signal!); + }) as typeof globalThis.fetch; + return { fetch, urls }; +} +const isWatch = (url: string) => new URL(url, "http://host").searchParams.get("watch") === "1"; +/** A step that never answers until the request is aborted, like a slow server. */ +const hang: Step = (_url, signal) => + new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(new DOMException("aborted", "AbortError")), { once: true }); + }); + +function statuses(handle: ResourceStreamHandle): ConnectionStatus[] { + const out: ConnectionStatus[] = []; + handle.subscribe((state) => out.push(state.status)); + return out; +} +const until = (handle: ResourceStreamHandle, status: ConnectionStatus) => + new Promise((resolve) => { + const off = handle.subscribe((state) => { + if (state.status === status) { + off(); + resolve(); + } + }); + }); +const pendingTimeouts = () => process.getActiveResourcesInfo().filter((r) => r === "Timeout").length; + +const url = "/k8s/api/v1/namespaces/app/configmaps?labelSelector=tier%3Dweb"; + +test("an empty collection is a complete snapshot, and the watch opens from its resourceVersion", async () => { + const { fetch, urls } = scripted( + () => collection([], "42"), + () => body("", { open: true }).response, + ); + const events: ResourceStateEvent[] = []; + const handle = connectNativeWatch(url, (event) => events.push(event), { fetch }); + await until(handle, "live"); + handle.close(); + await handle.closed; + assert.deepEqual(events, [{ type: "reset" }, { type: "synced" }]); + assert.equal(urls[0], url, "the LIST is the collection URL, with no limit or continue"); + const watch = new URL(urls[1]!, "http://host"); + assert.equal(watch.pathname, "/k8s/api/v1/namespaces/app/configmaps"); + assert.equal(watch.searchParams.get("labelSelector"), "tier=web", "the WATCH uses the same selectors"); + assert.equal(watch.searchParams.get("watch"), "1"); + assert.equal(watch.searchParams.get("resourceVersion"), "42"); + assert.equal(watch.searchParams.has("limit"), false); +}); + +test("requests use same-origin credentials, no cache, cancellation and the caller's headers", async () => { + const inits: RequestInit[] = []; + const fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + inits.push(init!); + return inits.length === 1 ? collection([]) : body("").response; + }) as typeof globalThis.fetch; + const handle = connectNativeWatch(url, ignore, { fetch, maxRetries: 0, headers: { "X-Host": "1" } }); + await handle.closed; + assert.equal(inits.length, 2); + for (const init of inits) { + assert.equal(init.credentials, "same-origin"); + assert.equal(init.cache, "no-store"); + assert.ok(init.signal instanceof AbortSignal); + assert.deepEqual(init.headers, { Accept: "application/json", "X-Host": "1" }); + } +}); + +test("the snapshot is applied, then the watch is accepted, and only then is the stream live", async () => { + const log: string[] = []; + let releaseWatch!: () => void; + const watchOpened = new Promise((resolve) => { + releaseWatch = resolve; + }); + const { fetch } = scripted( + () => collection([cm("a", "one", "1"), cm("b", "two", "2")], "5"), + async () => { + log.push("watch requested"); + await watchOpened; + log.push("watch accepted"); + return body("", { open: true }).response; + }, + ); + const handle = connectNativeWatch(url, (event) => log.push(`apply ${event.type}`), { fetch }); + handle.subscribe((state) => { + log.push(`state ${state.status}`); + if (state.status === "live") handle.close(); + }); + await new Promise((resolve) => setTimeout(resolve, 10)); + assert.equal(handle.state.status, "syncing", "still syncing while the watch is opening"); + releaseWatch(); + await handle.closed; + assert.deepEqual(log, [ + "state connecting", + "state syncing", + "state syncing", + "apply reset", + "apply added", + "apply added", + "watch requested", + "watch accepted", + "apply synced", + "state live", + "state closed", + ]); +}); + +test("the healthy period does not start while the watch is opening", async () => { + // One failed attempt, then a watch that takes longer than the healthy period to be refused. Had + // the timer started at the snapshot, the budget would be replenished and a third attempt made. + const { fetch, urls } = scripted( + () => json(status(503, "busy"), 503), + () => collection([]), + async (u) => { + assert.ok(isWatch(u)); + await new Promise((resolve) => setTimeout(resolve, 30)); + return json(status(503, "busy"), 503); + }, + ); + const states: Readonly[] = []; + const handle = connectNativeWatch(url, ignore, { fetch, retryDelayMs: 0, maxRetries: 2, healthyResetMs: 5 }); + handle.subscribe((state) => states.push(state)); + await handle.closed; + assert.equal(urls.length, 4, "two retries, then the scripted network failure ends the budget"); + assert.ok(!states.some((s) => s.status === "live")); + assert.deepEqual( + states.filter((s) => s.status === "retrying").map((s) => s.retries), + [1, 2], + ); + assert.equal(handle.state.status, "exhausted"); +}); + +test("missing item type metadata comes from a typed collection; present metadata and fields are kept", async () => { + const bare = { metadata: { uid: "a", name: "one", resourceVersion: "1" }, data: { k: "v" }, extra: [1] }; + const own = { apiVersion: "example.com/v2", kind: "Override", metadata: { uid: "b", name: "two" } }; + const half = { apiVersion: "v1", metadata: { uid: "c", name: "three" } }; + const { fetch } = scripted( + () => collection([bare, own, half], "9"), + () => + body( + frames({ type: "MODIFIED", object: { metadata: { uid: "a", name: "one", resourceVersion: "11" }, data: {} } }), + { open: true }, + ).response, + ); + const events: ResourceStateEvent[] = []; + const handle = connectNativeWatch(url, (event) => events.push(event), { fetch }); + await until(handle, "live"); + await new Promise((resolve) => setTimeout(resolve, 5)); + handle.close(); + await handle.closed; + const objects = events.flatMap((e) => ("object" in e ? [e.object] : [])); + assert.deepEqual(objects[0], { ...bare, apiVersion: "v1", kind: "ConfigMap" }); + assert.deepEqual(objects[1], own, "type metadata already present is never replaced"); + assert.deepEqual(objects[2], { ...half, kind: "ConfigMap" }); + assert.equal(objects[3]!.kind, "ConfigMap", "a watch object lacking its type gets the collection's item type"); + assert.ok(!objects.some((o) => o.kind === "ConfigMapList"), "never the collection's own kind"); +}); + +for (const [label, list] of [ + ["a generic List", { kind: "List", apiVersion: "v1" }], + ["a meta.k8s.io collection", { kind: "PartialObjectMetadataList", apiVersion: "meta.k8s.io/v1" }], +] as const) + test(`${label} does not name its items' type, so an untyped item is malformed`, async () => { + const errors: unknown[][] = []; + const events: ResourceStateEvent[] = []; + const { fetch, urls } = scripted(() => + collection([{ metadata: { uid: "a", name: "one" } }], "1", { kind: list.kind, apiVersion: list.apiVersion }), + ); + const handle = connectNativeWatch("/k8s/api/v1/configmaps", (event) => events.push(event), { + fetch, + maxRetries: 1, + retryDelayMs: 0, + onError: (...args) => errors.push(args), + }); + await handle.closed; + assert.deepEqual(events, [], "nothing from a collection it cannot type"); + assert.equal(urls.length, 2, "a malformed collection is retried within the budget"); + assert.equal(errors[0]![0], "INTERNAL"); + assert.match(String(errors[0]![1]), /native list: .*no apiVersion or kind/); + assert.equal(errors[0]![2], false); + assert.equal(handle.state.status, "exhausted"); + }); + +for (const [label, items, pattern] of [ + ["an item without a uid", [{ apiVersion: "v1", kind: "ConfigMap", metadata: { name: "x" } }], /no metadata\.uid/], + ["an item without a name", [{ apiVersion: "v1", kind: "ConfigMap", metadata: { uid: "x" } }], /no metadata\.name/], + ["a uid listed twice", [cm("a", "one", "1"), cm("a", "two", "2")], /UID a twice/], + ["an item that is not an object", ["nope"], /no metadata/], +] as const) + test(`${label} rejects the whole collection before any reset`, async () => { + const errors: unknown[][] = []; + const events: ResourceStateEvent[] = []; + const { fetch } = scripted(() => collection([...items])); + const handle = connectNativeWatch(url, (event) => events.push(event), { + fetch, + maxRetries: 0, + onError: (...args) => errors.push(args), + }); + await handle.closed; + assert.deepEqual(events, []); + assert.equal(errors.length, 1); + assert.match(String(errors[0]![1]), pattern); + }); + +for (const [label, response, pattern] of [ + ["no resourceVersion", json({ kind: "ConfigMapList", apiVersion: "v1", metadata: {}, items: [] }), /resourceVersion/], + ["no items", json({ kind: "ConfigMapList", apiVersion: "v1", metadata: { resourceVersion: "1" } }), /items/], + ["a body that is not JSON", new Response("login"), /not JSON/], +] as const) + test(`a collection with ${label} is malformed and retried`, async () => { + const errors: unknown[][] = []; + const { fetch } = scripted(() => response); + const handle = connectNativeWatch(url, ignore, { fetch, maxRetries: 0, onError: (...args) => errors.push(args) }); + await handle.closed; + assert.match(String(errors[0]![1]), pattern); + assert.equal(errors[0]![2], false); + assert.equal(handle.state.status, "exhausted"); + }); + +test("a paginated collection is refused, terminally, before reset: a page never establishes membership", async () => { + const store = new LiveResourceStore(readOnlyPolicy); + store.applyServerEvent(cm("old", "kept", "1")); + const errors: unknown[][] = []; + const { fetch, urls } = scripted(() => collection([cm("a", "one", "2")], "5", { continue: "token" })); + const handle = connectNativeWatch(url, (event) => applyStreamEvent(store, event), { + fetch, + retryDelayMs: 0, + onError: (...args) => errors.push(args), + }); + const states = statuses(handle); + await handle.closed; + assert.equal(urls.length, 1, "never retried and never watched"); + assert.ok(!urls[0]!.includes("limit") && !urls[0]!.includes("continue"), "no page was requested"); + assert.deepEqual(store.ids(), ["old"], "nothing applied and nothing pruned"); + assert.deepEqual(states, ["connecting", "syncing", "terminal"]); + assert.equal(errors.length, 1); + assert.equal(errors[0]![0], "INTERNAL"); + assert.match(String(errors[0]![1]), /paginated/); + assert.equal(errors[0]![2], true); +}); + +test("watch frames split anywhere — across lines, JSON and UTF-8 — and many per chunk apply in order", async () => { + const text = frames( + { type: "ADDED", object: cm("b", "two", "11", "héllo ✓ 🚀") }, + { type: "MODIFIED", object: cm("a", "one", "12", "ünïcödé") }, + { type: "BOOKMARK", object: { kind: "ConfigMap", apiVersion: "v1", metadata: { resourceVersion: "13" } } }, + { type: "DELETED", object: cm("b", "two", "14", "héllo ✓ 🚀") }, + ); + for (const chunk of [1, 3, 7, 1_000_000]) { + const store = new LiveResourceStore(readOnlyPolicy); + const types: string[] = []; + const watch = body(text, { chunk, open: true }); + const { fetch } = scripted( + () => collection([cm("a", "one", "10")], "10"), + () => watch.response, + ); + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + if (event.type === "deleted") handle.close(); + }, + { fetch }, + ); + await handle.closed; + assert.deepEqual(types, ["reset", "added", "synced", "added", "modified", "deleted"], `chunk ${chunk}`); + assert.deepEqual(store.ids(), ["a"]); + assert.deepEqual(store.server("a").data, { value: "ünïcödé" }, `chunk ${chunk}`); + assert.equal(store.server("a").metadata.resourceVersion, "12", "a bookmark changes no object"); + assert.equal(watch.state.cancelled, true); + } +}); + +test("a deletion carries the identity of the deleted object", async () => { + const events: ResourceStateEvent[] = []; + const clusterScoped = { apiVersion: "v1", kind: "Namespace", metadata: { uid: "ns", name: "app" } }; + const { fetch } = scripted( + () => collection([cm("a", "one", "1"), clusterScoped], "2", { kind: "List" }), + () => + body(frames({ type: "DELETED", object: cm("a", "one", "3") }, { type: "DELETED", object: clusterScoped })) + .response, + ); + const handle = connectNativeWatch(url, (event) => events.push(event), { fetch, maxRetries: 0 }); + await handle.closed; + assert.deepEqual( + events.filter((e) => e.type === "deleted"), + [ + { + type: "deleted", + identity: { uid: "a", apiVersion: "v1", kind: "ConfigMap", namespace: "app", name: "one" }, + }, + { type: "deleted", identity: { uid: "ns", apiVersion: "v1", kind: "Namespace", name: "app" } }, + ], + ); +}); + +test("a same-name recreation is a different resource: the old UID leaves and the new one arrives", async () => { + const store = new LiveResourceStore(readOnlyPolicy); + const { fetch } = scripted( + () => collection([cm("old", "same", "1", "first")]), + () => + body( + frames( + { type: "DELETED", object: cm("old", "same", "2", "first") }, + { type: "ADDED", object: cm("new", "same", "3", "second") }, + ), + ).response, + ); + const handle = connectNativeWatch(url, (event) => applyStreamEvent(store, event), { fetch, maxRetries: 0 }); + await handle.closed; + assert.deepEqual(store.ids(), ["new"]); + assert.deepEqual(store.server("new").data, { value: "second" }); +}); + +for (const [label, tail, pattern] of [ + ["a frame that is not JSON", "{not json}\n", /not JSON/], + ["a frame without an object", `${JSON.stringify({ type: "ADDED" })}\n`, /no type or object/], + [ + "an unknown event type", + `${JSON.stringify({ type: "RENAMED", object: cm("x", "x", "1") })}\n`, + /unknown watch event/, + ], + ["an object without a uid", `${JSON.stringify({ type: "ADDED", object: { metadata: { name: "x" } } })}\n`, /uid/], + ["a truncated frame at EOF", '{"type":"MODIFIED","object":{"metad', /ended inside a frame/], + ["invalid UTF-8", new Uint8Array([0xff, 0xfe, 0x0a]), /UTF-8/], + ["a character cut off at EOF", new Uint8Array([0xe2, 0x9c]), /UTF-8/], +] as const) + test(`${label} is reported, never skipped, and recovered by a fresh list`, async () => { + const errors: unknown[][] = []; + const types: string[] = []; + const store = new LiveResourceStore(readOnlyPolicy); + const first = body(frames({ type: "MODIFIED", object: cm("a", "one", "2", "before") }), { open: true }); + first.push(tail); + first.end(); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")]), + () => first.response, + () => collection([cm("a", "one", "3", "relisted")]), + () => body("", { open: true }).response, + ); + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + }, + { fetch, retryDelayMs: 0, onError: (...args) => errors.push(args) }, + ); + const states = statuses(handle); + let lives = 0; + handle.subscribe((state) => { + if (state.status === "live" && ++lives === 2) handle.close(); + }); + await handle.closed; + assert.equal(urls.length, 4); + assert.deepEqual(types, ["reset", "added", "synced", "modified", "reset", "added", "synced"]); + assert.deepEqual(store.server("a").data, { value: "relisted" }); + assert.equal(errors.length, 1); + assert.equal(errors[0]![0], "INTERNAL"); + assert.match(String(errors[0]![1]), /^native watch: /); + assert.match(String(errors[0]![1]), pattern); + assert.equal(errors[0]![2], false); + assert.ok(states.includes("retrying")); + }); + +test("HTTP 410 on the watch and an in-stream 410 both recover with a fresh snapshot", async () => { + const errors: unknown[][] = []; + const store = new LiveResourceStore(readOnlyPolicy); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1"), cm("b", "two", "1")], "1"), + () => json(status(410, "too old resource version: 1 (7)", { reason: "Expired" }), 410), + () => collection([cm("a", "one", "7")], "7"), + () => + body(frames({ type: "ERROR", object: status(410, "too old resource version: 7 (9)", { reason: "Gone" }) })) + .response, + () => collection([cm("a", "one", "9")], "9"), + () => body("", { open: true }).response, + ); + const handle = connectNativeWatch(url, (event) => applyStreamEvent(store, event), { + fetch, + retryDelayMs: 0, + onError: (...args) => errors.push(args), + }); + const states = statuses(handle); + let lives = 0; + handle.subscribe((state) => { + if (state.status === "live" && ++lives === 2) handle.close(); + }); + await handle.closed; + assert.equal(urls.length, 6); + assert.deepEqual(errors, [ + ["RESYNC_REQUIRED", "too old resource version: 1 (7)", false, undefined], + ["RESYNC_REQUIRED", "too old resource version: 7 (9)", false, undefined], + ]); + assert.deepEqual(store.ids(), ["a"], "b was pruned by the completed replacement snapshot"); + assert.equal(store.server("a").metadata.resourceVersion, "9"); + assert.deepEqual(new URL(urls[3]!, "http://host").searchParams.get("resourceVersion"), "7"); + assert.equal(states.filter((s) => s === "retrying").length, 2); +}); + +test("repeated expiry consumes the bounded budget with backoff instead of looping on fresh lists", async () => { + let lists = 0; + const waits: number[] = []; + const fetch = (async (input: RequestInfo | URL) => { + if (!isWatch(String(input))) { + lists++; + return collection([], String(lists)); + } + return json(status(410, "expired", { reason: "Expired" }), 410); + }) as typeof globalThis.fetch; + const handle = connectNativeWatch(url, ignore, { fetch, maxRetries: 3, retryDelayMs: 1, maxRetryDelayMs: 4 }); + handle.subscribe((state) => { + if (state.status === "retrying") waits.push(state.retryInMs!); + }); + await handle.closed; + assert.equal(lists, 4, "one list per attempt, and the budget ends it"); + assert.equal(waits.length, 3); + assert.equal(handle.state.status, "exhausted"); +}); + +test("short EOFs on an accepted watch also exhaust the budget", async () => { + let calls = 0; + const fetch = (async (input: RequestInfo | URL) => { + calls++; + return isWatch(String(input)) ? body("").response : collection([]); + }) as typeof globalThis.fetch; + const handle = connectNativeWatch(url, ignore, { fetch, maxRetries: 2, retryDelayMs: 0, healthyResetMs: 60_000 }); + const states = statuses(handle); + await handle.closed; + assert.equal(calls, 6); + assert.equal(states.filter((s) => s === "live").length, 3); + assert.equal(handle.state.status, "exhausted"); +}); + +for (const [phase, code, expected] of [ + ["list", 401, "UNAUTHENTICATED"], + ["list", 403, "FORBIDDEN"], + ["watch", 401, "UNAUTHENTICATED"], + ["watch", 403, "FORBIDDEN"], + ["list", 404, "INTERNAL"], + ["watch", 400, "INTERNAL"], +] as const) + test(`HTTP ${code} on the ${phase} is terminal and keeps the Status message`, async () => { + const errors: unknown[][] = []; + const refusal = () => json(status(code, `${phase} refused with ${code}`), code); + const { fetch, urls } = scripted(...(phase === "list" ? [refusal] : [() => collection([]), refusal])); + const handle = connectNativeWatch(url, ignore, { fetch, retryDelayMs: 0, onError: (...args) => errors.push(args) }); + await handle.closed; + assert.equal(urls.length, phase === "list" ? 1 : 2, "never retried"); + assert.deepEqual(errors, [[expected, `${phase} refused with ${code}`, true, undefined]]); + assert.equal(handle.state.status, "terminal"); + }); + +test("an in-stream 403 is terminal and nothing after it is applied", async () => { + const errors: unknown[][] = []; + const types: string[] = []; + const { fetch, urls } = scripted( + () => collection([]), + () => + body( + frames( + { type: "ERROR", object: status(403, "access revoked", { reason: "Forbidden" }) }, + { type: "ADDED", object: cm("a", "one", "2") }, + ), + { open: true }, + ).response, + ); + const handle = connectNativeWatch(url, (event) => types.push(event.type), { + fetch, + onError: (...args) => errors.push(args), + }); + await handle.closed; + assert.equal(urls.length, 2); + assert.deepEqual(types, ["reset", "synced"]); + assert.deepEqual(errors, [["FORBIDDEN", "access revoked", true, undefined]]); + assert.equal(handle.state.status, "terminal"); +}); + +test("a refusal without a Status body is still classified, with a native message", async () => { + const errors: unknown[][] = []; + const { fetch } = scripted(() => new Response("denied", { status: 403 })); + const handle = connectNativeWatch(url, ignore, { fetch, onError: (...args) => errors.push(args) }); + await handle.closed; + assert.deepEqual(errors, [["FORBIDDEN", "native list: HTTP 403", true, undefined]]); +}); + +for (const code of [408, 429, 500, 503]) + test(`HTTP ${code} is retried`, async () => { + const errors: unknown[][] = []; + const { fetch, urls } = scripted( + () => json(status(code, "try later"), code), + () => collection([]), + () => json(status(code, "try later"), code), + ); + const handle = connectNativeWatch(url, ignore, { + fetch, + retryDelayMs: 0, + maxRetries: 2, + onError: (...args) => errors.push(args), + }); + await handle.closed; + assert.equal(urls.length, 4, "list, list + watch, then the scripted network failure"); + assert.equal(errors.length, 2); + assert.ok(errors.every((e) => e[2] === false)); + assert.equal(handle.state.status, "exhausted"); + }); + +test("Retry-After and a Status's retryAfterSeconds set the least the reconnect waits", async () => { + const cases: [Step[], unknown[]][] = [ + [ + [() => json(status(429, "slow down"), 429, { "Retry-After": "2" })], + ["UPSTREAM_UNAVAILABLE", "slow down", false, 2000], + ], + [ + [ + () => collection([]), + () => + body( + frames({ + type: "ERROR", + object: status(500, "etcd leader changed", { + reason: "InternalError", + details: { retryAfterSeconds: 3 }, + }), + }), + ).response, + ], + ["INTERNAL", "etcd leader changed", false, 3000], + ], + ]; + for (const [steps, error] of cases) { + const errors: unknown[][] = []; + const waits: number[] = []; + const handle = connectNativeWatch(url, ignore, { + fetch: scripted(...steps).fetch, + retryDelayMs: 0, + maxRetryDelayMs: 10_000, + onError: (...args) => errors.push(args), + }); + handle.subscribe((state) => { + if (state.status !== "retrying") return; + waits.push(state.retryInMs!); + handle.close(); + }); + await handle.closed; + assert.deepEqual(errors, [error]); + assert.deepEqual(waits, [error[3]]); + } +}); + +test("network failures on list, watch and body read consume exactly the retry budget", async () => { + let calls = 0; + const failing = new ReadableStream({ + pull(controller) { + controller.error(new TypeError("connection reset")); + }, + }); + const steps: Step[] = [ + () => { + throw new TypeError("offline"); + }, + () => collection([]), + () => { + throw new TypeError("offline"); + }, + () => collection([]), + () => new Response(failing), + ]; + const errors: unknown[][] = []; + const fetch = (async () => steps[calls++]!("", new AbortController().signal)) as typeof globalThis.fetch; + const handle = connectNativeWatch(url, ignore, { + fetch, + retryDelayMs: 0, + maxRetries: 2, + onError: (...args) => errors.push(args), + }); + await handle.closed; + assert.equal(calls, 5); + assert.deepEqual(errors, [], "a network failure is not malformed input"); + assert.equal(handle.state.status, "exhausted"); +}); + +test("missed deletes and selector exits are repaired by re-listing; prior state is kept until then", async () => { + const store = new LiveResourceStore(readOnlyPolicy); + const { fetch } = scripted( + // The first connection sees a, b and c; its watch then ends without reporting b's deletion or + // c's label change taking it out of the selector. + () => collection([cm("a", "one", "1"), cm("b", "two", "1"), cm("c", "three", "1")], "1"), + () => body("").response, + // The replacement snapshot no longer lists them, but its watch is refused: incomplete. + () => collection([cm("a", "one", "5")], "5"), + () => json(status(503, "busy"), 503), + () => collection([cm("a", "one", "6")], "6"), + () => body("", { open: true }).response, + ); + const seen: string[][] = []; + const handle = connectNativeWatch( + url, + (event) => { + applyStreamEvent(store, event); + if (event.type === "synced") seen.push(store.ids().sort()); + }, + { fetch, retryDelayMs: 0 }, + ); + handle.subscribe((state) => { + if (state.status === "retrying") seen.push(store.ids().sort()); + }); + let lives = 0; + handle.subscribe((state) => { + if (state.status === "live" && ++lives === 2) handle.close(); + }); + await handle.closed; + assert.deepEqual(seen, [["a", "b", "c"], ["a", "b", "c"], ["a", "b", "c"], ["a"]]); + assert.equal(store.server("a").metadata.resourceVersion, "6"); +}); + +test("close during the list request, the list body, the watch request, the watch body and backoff", async (t) => { + for (const at of ["list fetch", "list body", "watch fetch", "watch body", "backoff"] as const) { + await t.test(at, async () => { + const before = pendingTimeouts(); + const listBody = body(JSON.stringify({ kind: "ConfigMapList", apiVersion: "v1" }).slice(0, 10), { open: true }); + const watchBody = body("", { open: true }); + const steps: Step[] = + at === "list fetch" + ? [hang] + : at === "list body" + ? [() => listBody.response] + : at === "backoff" + ? [() => json(status(503, "busy"), 503)] + : [() => collection([cm("a", "one", "1")]), at === "watch fetch" ? hang : () => watchBody.response]; + const { fetch, urls } = scripted(...steps); + const events: string[] = []; + const handle = connectNativeWatch(url, (event) => events.push(event.type), { + fetch, + retryDelayMs: 60_000, + healthyResetMs: 60_000, + }); + const states = statuses(handle); + const trigger: ConnectionStatus = + at === "backoff" ? "retrying" : at === "watch body" ? "live" : at === "list fetch" ? "connecting" : "syncing"; + await until(handle, trigger); + // Let the request or read get under way, so close() lands while it is pending. + if (at === "watch body") watchBody.push(frames({ type: "ADDED", object: cm("b", "two", "2") })); + await new Promise((resolve) => setTimeout(resolve, 5)); + handle.close(); + await handle.closed; + const count = events.length; + if (at === "watch body") { + // The connector cancelled the body, so a late frame cannot even be enqueued. + assert.throws(() => watchBody.push(frames({ type: "MODIFIED", object: cm("b", "two", "3") }))); + } + await new Promise((resolve) => setTimeout(resolve, 5)); + assert.equal(events.length, count, "nothing is delivered after close"); + assert.equal(states.at(-1), "closed"); + assert.equal(states.filter((s) => s === "live").length, at === "watch body" ? 1 : 0); + assert.equal(urls.length, steps.length, "nothing is requested after close"); + if (at === "list body") assert.equal(listBody.state.cancelled, true); + if (at === "watch body") { + assert.equal(watchBody.state.cancelled, true); + assert.deepEqual(events, ["reset", "added", "synced", "added"]); + } + if (at === "watch fetch") assert.deepEqual(events, ["reset", "added"], "no synced without an accepted watch"); + assert.equal(pendingTimeouts(), before, "the backoff and health timers are released"); + }); + } +}); + +test("an already aborted signal never fetches; an external abort ends a live watch", async () => { + const { fetch, urls } = scripted(); + const aborted = connectNativeWatch(url, ignore, { fetch, signal: AbortSignal.abort() }); + await aborted.closed; + assert.equal(urls.length, 0); + assert.equal(aborted.state.status, "closed"); + + const controller = new AbortController(); + const watch = body("", { open: true }); + const live = connectNativeWatch(url, ignore, { + signal: controller.signal, + fetch: scripted( + () => collection([]), + () => watch.response, + ).fetch, + }); + await until(live, "live"); + controller.abort(); + await live.closed; + assert.equal(watch.state.cancelled, true); + assert.equal(live.state.status, "closed"); +}); + +test("a consumer exception stops the stream, cancels the watch and rejects closed with it", async () => { + const boom = new Error("render failed"); + const watch = body( + frames({ type: "MODIFIED", object: cm("a", "one", "2") }, { type: "MODIFIED", object: cm("a", "one", "3") }), + { open: true }, + ); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")]), + () => watch.response, + ); + const seen: string[] = []; + const errors: unknown[][] = []; + const before = pendingTimeouts(); + const handle = connectNativeWatch( + url, + (event) => { + seen.push(event.type); + if (event.type === "modified") throw boom; + }, + { fetch, retryDelayMs: 0, healthyResetMs: 60_000, onError: (...args) => errors.push(args) }, + ); + const states = statuses(handle); + await assert.rejects(handle.closed, (error) => error === boom); + assert.equal(urls.length, 2, "never retried"); + assert.deepEqual(seen, ["reset", "added", "synced", "modified"]); + assert.deepEqual(errors, []); + assert.deepEqual(states, ["connecting", "syncing", "syncing", "live", "closed"]); + assert.equal(watch.state.cancelled, true); + assert.equal(pendingTimeouts(), before); +}); + +test("a consumer exception during the snapshot never opens the watch", async () => { + const boom = new Error("bad row"); + const { fetch, urls } = scripted(() => collection([cm("a", "one", "1"), cm("b", "two", "1")])); + const seen: string[] = []; + const handle = connectNativeWatch( + url, + (event) => { + seen.push(event.type); + if (event.type === "added") throw boom; + }, + { fetch }, + ); + await assert.rejects(handle.closed, (error) => error === boom); + assert.deepEqual(seen, ["reset", "added"]); + assert.equal(urls.length, 1); + assert.equal(handle.state.status, "closed"); +}); + +test("a consumer that closes while applying synced gets no live", async () => { + const watch = body("", { open: true }); + const { fetch } = scripted( + () => collection([]), + () => watch.response, + ); + const handle = connectNativeWatch( + url, + (event) => { + if (event.type === "synced") handle.close(); + }, + { fetch }, + ); + const states = statuses(handle); + await handle.closed; + assert.deepEqual(states, ["connecting", "syncing", "syncing", "closed"]); + assert.equal(watch.state.cancelled, true); +}); + +test("a subscriber or onError that throws stops the stream without retrying", async () => { + const boom = new Error("host bug"); + const subscriber = connectNativeWatch(url, ignore, { + retryDelayMs: 0, + fetch: scripted( + () => collection([]), + () => body("", { open: true }).response, + ).fetch, + }); + subscriber.subscribe((state) => { + if (state.status === "live") throw boom; + }); + await assert.rejects(subscriber.closed, (error) => error === boom); + assert.equal(subscriber.state.status, "closed"); + + const { fetch, urls } = scripted(() => json(status(503, "busy"), 503)); + const reporter = connectNativeWatch(url, ignore, { + fetch, + retryDelayMs: 0, + onError: () => { + throw boom; + }, + }); + await assert.rejects(reporter.closed, (error) => error === boom); + assert.equal(urls.length, 1); + assert.equal(reporter.state.status, "closed"); +}); + +test("the collection URL may not set what the connector sets", () => { + for (const param of ["watch=1", "resourceVersion=5", "limit=10", "continue=x", "resourceVersionMatch=Exact"]) { + assert.throws(() => connectNativeWatch(`/k8s/api/v1/configmaps?${param}`, ignore), /must not set/, param); + } +}); + +test("nativeCollectionURL addresses core and grouped, namespaced and cluster-scoped collections", () => { + assert.equal( + nativeCollectionURL("/k8s", { version: "v1", resource: "configmaps", namespace: "app" }), + "/k8s/api/v1/namespaces/app/configmaps", + ); + assert.equal( + nativeCollectionURL("/k8s/", { group: "apps", version: "v1", resource: "deployments" }), + "/k8s/apis/apps/v1/deployments", + ); + assert.equal( + nativeCollectionURL("", { group: "", version: "v1", resource: "namespaces", namespace: "" }), + "/api/v1/namespaces", + ); + assert.equal( + nativeCollectionURL("https://host.example/proxy", { + group: "widgets.example.com", + version: "v1alpha1", + resource: "widgets", + namespace: "team-a", + labelSelector: "app in (a,b),tier!=db", + name: "w 1+2", + }), + "https://host.example/proxy/apis/widgets.example.com/v1alpha1/namespaces/team-a/widgets" + + "?labelSelector=app+in+%28a%2Cb%29%2Ctier%21%3Ddb&fieldSelector=metadata.name%3Dw+1%2B2", + ); + const query = new URL( + nativeCollectionURL("/k8s", { version: "v1", resource: "pods", name: "web-0", labelSelector: "a=b" }), + "http://host", + ).searchParams; + assert.equal(query.get("labelSelector"), "a=b"); + assert.equal(query.get("fieldSelector"), "metadata.name=web-0"); +}); + +test("nativeCollectionURL refuses segments that would change the path", () => { + for (const scope of [ + { version: "v1", resource: "" }, + { version: "", resource: "pods" }, + { version: "v1", resource: "pods", namespace: ".." }, + { version: "v1", resource: "pods", namespace: "a/b" }, + { group: ".", version: "v1", resource: "pods" }, + { version: "v1", resource: "../secrets" }, + ]) { + assert.throws(() => nativeCollectionURL("/k8s", scope), /krm-stream: invalid/, JSON.stringify(scope)); + } + assert.equal( + nativeCollectionURL("/k8s", { version: "v1", resource: "pods", namespace: "a b" }), + "/k8s/api/v1/namespaces/a%20b/pods", + ); +}); From ed7a35ecd826382703237c74ab03414f9b5f5bf9 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 5 Oct 2026 11:27:24 +0000 Subject: [PATCH 3/6] docs(client): correct the native refusal helper's comment Co-Authored-By: Claude Opus 5.5 --- packages/krm-stream/src/native.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/krm-stream/src/native.ts b/packages/krm-stream/src/native.ts index 7e16326..e1b8f64 100644 --- a/packages/krm-stream/src/native.ts +++ b/packages/krm-stream/src/native.ts @@ -85,7 +85,7 @@ async function watchOnce( // Called detached, so a consumer never sees this hooks object as `this`. const { consume } = hooks; - /** Report a refused response; false when the connection must end. */ + /** Report a refused response. The connection ends either way. */ const refused = async (phase: string, res: Response) => { const { code, terminal } = classify(res.status); const message = (await statusMessage(res, controller.signal)) ?? `native ${phase}: HTTP ${res.status}`; From 6e671cbe1a9ca0d23b67a67c496c02902a0cbadd Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 5 Oct 2026 11:27:24 +0000 Subject: [PATCH 4/6] test(examples): add a native viewer with browser and real-API checks examples/native-viewer is one HTML page that lists and watches a native collection through a host proxy with connectNativeWatch and a read-only store, showing identity, fields, connection state and disposal. The Chromium suite loads it on both entry points with Playwright playing the proxy: snapshot before live, re-list pruning, disconnect and a terminal Status refusal. TestRealAPINativeWatchThroughHostProxy runs the connector in node through a credential-holding /k8s reverse proxy against a real API server: typed-list type metadata, modification, deletion, selector exit and same-name recreation. Co-Authored-By: Claude Opus 5.5 --- examples/native-viewer/README.md | 42 +++ examples/native-viewer/index.html | 117 ++++++++ .../tests/native-viewer.spec.ts | 144 ++++++++++ gateway/kube/native_e2e_test.go | 269 ++++++++++++++++++ packages/krm-stream/e2e/native-driver.ts | 43 +++ 5 files changed, 615 insertions(+) create mode 100644 examples/native-viewer/README.md create mode 100644 examples/native-viewer/index.html create mode 100644 examples/vanilla-browser/tests/native-viewer.spec.ts create mode 100644 gateway/kube/native_e2e_test.go create mode 100644 packages/krm-stream/e2e/native-driver.ts diff --git a/examples/native-viewer/README.md b/examples/native-viewer/README.md new file mode 100644 index 0000000..ee91263 --- /dev/null +++ b/examples/native-viewer/README.md @@ -0,0 +1,42 @@ +# Native viewer example + +A read-only live list of native Kubernetes resources, read through a proxy the host already runs. +One HTML page with no framework, bundler, gateway, editor or write endpoint. It uses +`connectNativeWatch` and `nativeCollectionURL` with `LiveResourceStore(readOnlyPolicy)`, and shows each +resource's identity (namespace, name, kind, UID, resourceVersion), its current fields and the +connection state. **Disconnect**, or leaving the page, closes the connection. + +## Run it against a cluster + +Build the library, then let `kubectl proxy` act as the host: it serves the API server under `/k8s` +with your kubeconfig credentials and serves this repository's files from the same origin. + +```bash +task build-client +kubectl proxy --port=8001 --api-prefix=/k8s/ --www=. --www-prefix=/files/ +# http://127.0.0.1:8001/files/examples/native-viewer/?namespace=default +``` + +Change a ConfigMap in that namespace and the row follows. Query parameters choose the collection: +`proxy` (default `/k8s`), `group`, `version` (default `v1`), `resource` (default `configmaps`), +`namespace` (default `default`; empty for cluster scope), `labelSelector` and `name`. `entry=bundle` +loads the single-file build. + +`kubectl proxy` is a development stand-in. A real host serves its own authenticated proxy, such as +krm-foyer's session-authenticated `/k8s`, and decides which collections each user may read. The +browser never holds a Kubernetes credential. + +## What it shows and what it does not + +- The page lists the collection, applies the complete snapshot and becomes `live` only once the + watch is accepted. The **History** line shows each connection state. +- Every reconnect lists again; a resource deleted or relabelled out of the selector while + disconnected disappears when the next snapshot completes. +- Objects are exactly what the proxy returns, Secret values and machinery fields included. Native + access provides no projection, redaction, suppression or shared watches; use the + [gateway](../../docs/why-a-gateway.md) for those. +- It is a viewer. Native editing, resumable watches and paginated lists are later work; see the + [delivery plan](../../docs/proposals/0006-stream-and-save-implementation-plan.md#open-work-and-delivery-order). + +`task e2e-browser` loads this page in Chromium on both entry points, with Playwright standing in for +the host proxy: [native-viewer.spec.ts](../vanilla-browser/tests/native-viewer.spec.ts). diff --git a/examples/native-viewer/index.html b/examples/native-viewer/index.html new file mode 100644 index 0000000..4141f1a --- /dev/null +++ b/examples/native-viewer/index.html @@ -0,0 +1,117 @@ + + + + + + + krm-stream — native viewer + + + +

Native viewer

+

+

+ State: loading + +

+

+

History:

+ + + +
NameKindUIDresourceVersionFields
+ + + + diff --git a/examples/vanilla-browser/tests/native-viewer.spec.ts b/examples/vanilla-browser/tests/native-viewer.spec.ts new file mode 100644 index 0000000..89f3f55 --- /dev/null +++ b/examples/vanilla-browser/tests/native-viewer.spec.ts @@ -0,0 +1,144 @@ +// The native viewer example (examples/native-viewer) in a real browser, on both entry points. +// +// No cluster: Playwright plays the host's /k8s proxy on the page's own origin, answering each LIST +// and WATCH itself, so the test decides when a watch is accepted and what it says. It also serves the +// example page and the built library from the repository, where the page imports them by relative +// path, exactly as `kubectl proxy --www` does in the example's README. + +import { readFile } from "node:fs/promises"; +import { extname } from "node:path"; +import { fileURLToPath } from "node:url"; +import type { Page } from "@playwright/test"; +import { expect, test } from "./fixtures.ts"; + +const repository = fileURLToPath(new URL("../../../", import.meta.url)); +const types: Record = { ".html": "text/html", ".js": "text/javascript" }; + +async function serveFromRepository(page: Page) { + for (const prefix of ["/examples/native-viewer/", "/packages/krm-stream/dist/"]) { + await page.route(`**${prefix}**`, async (route) => { + const path = new URL(route.request().url()).pathname; + await route.fulfill({ + body: await readFile(`${repository}${path.slice(1)}`), + contentType: types[extname(path)] ?? "application/octet-stream", + }); + }); + } +} + +/** A ConfigMap as a typed LIST returns it: no apiVersion or kind on the item. */ +const item = (name: string, uid: string, rv: string, value: string) => ({ + metadata: { name, uid, namespace: "app", resourceVersion: rv, labels: { tier: "web" } }, + data: { value }, +}); +const typed = (object: ReturnType) => ({ apiVersion: "v1", kind: "ConfigMap", ...object }); +const frames = (...events: unknown[]) => events.map((e) => `${JSON.stringify(e)}\n`).join(""); + +function gate() { + let open!: () => void; + const opened = new Promise((resolve) => { + open = resolve; + }); + return { opened, open }; +} + +test("the native viewer lists, goes live once the watch is accepted, re-lists and disconnects", async ({ + page, + entry, +}) => { + const errors: string[] = []; + page.on("pageerror", (e) => errors.push(e.message)); + await serveFromRepository(page); + + const a1 = item("a", "uid-a", "1", "one"); + const a2 = item("a", "uid-a", "11", "two"); + const b = item("b", "uid-b", "1", "one"); + const c = item("c", "uid-c", "12", "new"); + const watches = [gate(), gate()]; + let lists = 0; + let watchCount = 0; + await page.route("**/k8s/**", async (route) => { + const url = new URL(route.request().url()); + expect(url.pathname).toBe("/k8s/api/v1/namespaces/app/configmaps"); + expect(url.searchParams.get("labelSelector"), "LIST and WATCH share the selectors").toBe("tier=web"); + if (url.searchParams.get("watch") !== "1") { + lists++; + // The first snapshot holds a and b. Every later one holds only a: c is deleted while the page + // is between watches, and nothing reports it except the next complete snapshot. + const items = lists === 1 ? [a1, b] : [a2]; + await route.fulfill({ + json: { kind: "ConfigMapList", apiVersion: "v1", metadata: { resourceVersion: `${lists}0` }, items }, + }); + return; + } + const n = watchCount++; + expect(url.searchParams.get("resourceVersion")).toBe(`${n + 1}0`); + const held = watches[n]; + if (!held) return; // a later watch is never answered: the page stays syncing until it disconnects + await held.opened; + // A fulfilled body ends, so each answered watch is followed by EOF and a fresh LIST. + await route.fulfill({ + contentType: "application/json", + body: + n === 0 + ? frames( + { type: "MODIFIED", object: typed(a2) }, + { type: "DELETED", object: typed(b) }, + { type: "ADDED", object: typed(c) }, + ) + : "", + }); + }); + + await page.goto(`/examples/native-viewer/index.html?entry=${entry}&namespace=app&labelSelector=tier%3Dweb`); + const row = (uid: string) => page.locator(`tr[data-uid="${uid}"]`); + const state = page.locator("#state"); + const history = page.locator("#history"); + + // The snapshot is applied, but the watch is still opening: not live yet. + await expect(row("uid-a")).toContainText("v1 ConfigMap"); + await expect(row("uid-b")).toBeVisible(); + await expect(state).toHaveText("syncing"); + await expect(history).not.toContainText("live"); + + watches[0]!.open(); + await expect(history).toContainText("syncing → live"); + // The watch's events arrived; then it ended, and the next LIST is incomplete until its watch is + // accepted, so c — missing from that LIST — is still shown. + await expect.poll(() => lists).toBe(2); + await expect(state).toHaveText("syncing"); + await expect(row("uid-a")).toContainText('"value": "two"'); + await expect(row("uid-b")).toHaveCount(0); + await expect(row("uid-c")).toBeVisible(); + + watches[1]!.open(); + // The replacement snapshot completes: c is pruned. + await expect(row("uid-c")).toHaveCount(0); + await expect(row("uid-a")).toBeVisible(); + await expect.poll(() => lists).toBe(3); + await expect(state).toHaveText("syncing"); + + await page.getByRole("button", { name: "Disconnect" }).click(); + await expect(state).toHaveText("closed"); + const requested = lists + watchCount; + await page.waitForTimeout(1_000); + expect(lists + watchCount, "nothing is requested after disconnect").toBe(requested); + await expect(page.locator("#error")).toHaveText(""); + expect(errors).toEqual([]); +}); + +test("a refused native list is terminal and shows the proxy's Status message", async ({ page, entry }) => { + await serveFromRepository(page); + let requests = 0; + await page.route("**/k8s/**", async (route) => { + requests++; + await route.fulfill({ + status: 403, + json: { kind: "Status", apiVersion: "v1", status: "Failure", code: 403, message: "configmaps is forbidden" }, + }); + }); + await page.goto(`/examples/native-viewer/index.html?entry=${entry}&namespace=app`); + await expect(page.locator("#state")).toHaveText("terminal"); + await expect(page.locator("#error")).toHaveText("FORBIDDEN (terminal): configmaps is forbidden"); + expect(requests).toBe(1); +}); diff --git a/gateway/kube/native_e2e_test.go b/gateway/kube/native_e2e_test.go new file mode 100644 index 0000000..5561ad9 --- /dev/null +++ b/gateway/kube/native_e2e_test.go @@ -0,0 +1,269 @@ +//go:build e2e + +// Native viewing against a real API server: the browser's connectNativeWatch, run by node +// (packages/krm-stream/e2e/native-driver.ts), through a host-style /k8s proxy that holds the +// credentials. No gateway is involved; this checks that a real LIST and WATCH — typed collections +// whose items carry no apiVersion or kind, selector exits, deletes and same-name recreation — reach +// the read-only store as the connector promises. +// +// task test-real-api +package kube_test + +import ( + "bufio" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "os" + "os/exec" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/rest" +) + +// nativeProxy is the host: it serves the API server's paths under /k8s with its own credentials, for +// GET only. The browser never holds a Kubernetes credential. +func nativeProxy(t *testing.T) *httptest.Server { + t.Helper() + cfg := restConfig(t) + transport, err := rest.TransportFor(cfg) + if err != nil { + t.Fatalf("transport: %v", err) + } + target, err := url.Parse(cfg.Host) + if err != nil { + t.Fatalf("API server address: %v", err) + } + proxy := &httputil.ReverseProxy{ + Rewrite: func(r *httputil.ProxyRequest) { + r.Out.URL.Scheme = target.Scheme + r.Out.URL.Host = target.Host + r.Out.Host = target.Host + r.Out.URL.Path = strings.TrimPrefix(r.In.URL.Path, "/k8s") + r.Out.URL.RawPath = "" + }, + Transport: transport, + // A watch is a long response; every frame must reach the browser as it is written. + FlushInterval: -1, + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || !strings.HasPrefix(r.URL.Path, "/k8s/") { + http.Error(w, "not proxied", http.StatusNotFound) + return + } + proxy.ServeHTTP(w, r) + })) + t.Cleanup(func() { + server.CloseClientConnections() + server.Close() + }) + return server +} + +type nativeMember struct { + UID string `json:"uid"` + Name string `json:"name"` + Kind string `json:"kind"` + Data map[string]string `json:"data"` +} + +// nativeLine is one line the driver prints: a connection state, a state event with what the store +// then holds, or an error. +type nativeLine struct { + State string `json:"state"` + Event string `json:"event"` + Members []nativeMember `json:"members"` + Error *struct { + Code string `json:"code"` + Message string `json:"message"` + Terminal bool `json:"terminal"` + } `json:"error"` +} + +type nativeDriver struct { + t *testing.T + stdin io.WriteCloser + lines chan nativeLine + members []nativeMember +} + +func newNativeDriver(t *testing.T, collectionURL string) *nativeDriver { + t.Helper() + node, err := exec.LookPath("node") + if err != nil { + t.Fatalf("node runs the browser connector in this test: %v", err) + } + cmd := exec.Command(node, filepath.Join("..", "..", "packages", "krm-stream", "e2e", "native-driver.ts"), collectionURL) //nolint:gosec // a fixed path in this repository + cmd.Stderr = os.Stderr + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + d := &nativeDriver{t: t, stdin: stdin, lines: make(chan nativeLine, 256)} + go func() { + defer close(d.lines) + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 1<<20), 16<<20) + for scanner.Scan() { + var line nativeLine + if json.Unmarshal(scanner.Bytes(), &line) == nil { + d.lines <- line + } + } + }() + t.Cleanup(func() { + _ = stdin.Close() + _ = cmd.Wait() + }) + return d +} + +// await reads lines until pred holds, failing on any error the connector reports. +func (d *nativeDriver) await(what string, pred func(nativeLine) bool) nativeLine { + d.t.Helper() + deadline := time.After(30 * time.Second) + for { + select { + case line, ok := <-d.lines: + if !ok { + d.t.Fatalf("%s: the driver exited", what) + } + if line.Error != nil { + d.t.Fatalf("%s: connector error %s: %s (terminal %v)", what, line.Error.Code, line.Error.Message, line.Error.Terminal) + } + if line.Event != "" { + d.members = line.Members + } + if pred(line) { + return line + } + case <-deadline: + d.t.Fatalf("%s: not reached; the store holds %+v", what, d.members) + } + } +} + +func (d *nativeDriver) names() []string { + names := []string{} + for _, m := range d.members { + names = append(names, m.Name) + } + return names +} + +func (d *nativeDriver) member(name string) nativeMember { + for _, m := range d.members { + if m.Name == name { + return m + } + } + d.t.Fatalf("%s is not in the store: %+v", name, d.members) + return nativeMember{} +} + +func TestRealAPINativeWatchThroughHostProxy(t *testing.T) { + cs, _ := clients(t) + ns := scratchNamespace(t, cs) + ctx := context.Background() + configMaps := cs.CoreV1().ConfigMaps(ns) + create := func(name, tier, value string) *corev1.ConfigMap { + t.Helper() + created, err := configMaps.Create(ctx, &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{"tier": tier}}, + Data: map[string]string{"value": value}, + }, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("create %s: %v", name, err) + } + return created + } + update := func(name string, change func(*corev1.ConfigMap)) { + t.Helper() + current, err := configMaps.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get %s: %v", name, err) + } + change(current) + if _, err := configMaps.Update(ctx, current, metav1.UpdateOptions{}); err != nil { + t.Fatalf("update %s: %v", name, err) + } + } + remove := func(name string) { + t.Helper() + if err := configMaps.Delete(ctx, name, metav1.DeleteOptions{}); err != nil { + t.Fatalf("delete %s: %v", name, err) + } + } + hasNames := func(want ...string) func(nativeLine) bool { + return func(line nativeLine) bool { + if line.Event == "" { + return false + } + names := []string{} + for _, m := range line.Members { + names = append(names, m.Name) + } + return slices.Equal(names, want) + } + } + + create("a", "web", "1") + create("b", "web", "1") + create("c", "db", "1") + + proxy := nativeProxy(t) + d := newNativeDriver(t, proxy.URL+"/k8s/api/v1/namespaces/"+url.PathEscape(ns)+"/configmaps?labelSelector=tier%3Dweb") + + d.await("live", func(line nativeLine) bool { return line.State == "live" }) + if got := d.names(); !slices.Equal(got, []string{"a", "b"}) { + t.Fatalf("snapshot: want the selector's members a and b, got %v", got) + } + if kind := d.member("a").Kind; kind != "ConfigMap" { + t.Fatalf("a typed list's items carry no kind; the connector must supply ConfigMap, got %q", kind) + } + + update("a", func(cm *corev1.ConfigMap) { cm.Data["value"] = "2" }) + d.await("a modified", func(line nativeLine) bool { + return line.Event == "modified" && len(line.Members) > 0 && line.Members[0].Data["value"] == "2" + }) + + remove("b") + d.await("b deleted", hasNames("a")) + + // A label change that takes a out of the selector arrives as a deletion. + update("a", func(cm *corev1.ConfigMap) { cm.Labels["tier"] = "db" }) + d.await("a left the selector", hasNames()) + + first := create("d", "web", "1") + d.await("d added", hasNames("d")) + remove("d") + second := create("d", "web", "2") + d.await("d recreated", func(line nativeLine) bool { + return hasNames("d")(line) && line.Members[0].UID == string(second.UID) + }) + if first.UID == second.UID { + t.Fatal("the recreated object kept its UID; the case proves nothing") + } + + if _, err := io.WriteString(d.stdin, "close\n"); err != nil { + t.Fatal(err) + } + d.await("closed", func(line nativeLine) bool { return line.State == "closed" }) +} diff --git a/packages/krm-stream/e2e/native-driver.ts b/packages/krm-stream/e2e/native-driver.ts new file mode 100644 index 0000000..b8c885f --- /dev/null +++ b/packages/krm-stream/e2e/native-driver.ts @@ -0,0 +1,43 @@ +// A native viewer on stdout, for the Go real-API suite (gateway/kube/native_e2e_test.go). +// +// The Go test owns the cluster and a host-style /k8s proxy that holds the credentials. This process is +// the browser: the REAL connectNativeWatch feeding the REAL read-only store, with no gateway in +// between. After every state event and connection state it prints one JSON line describing what the +// store holds, so the Go side can wait for the cluster's changes to arrive. A line `close` on stdin +// closes the connection; the driver exits once `closed` settles. +// +// node e2e/native-driver.ts + +import { createInterface } from "node:readline"; +import { applyStreamEvent, connectNativeWatch, LiveResourceStore, readOnlyPolicy } from "../src/index.ts"; + +const url = process.argv[2]; +if (!url) throw new Error("usage: native-driver.ts "); + +const store = new LiveResourceStore(readOnlyPolicy); +const report = (line: object) => process.stdout.write(`${JSON.stringify(line)}\n`); +const members = () => + store + .ids() + .map((uid) => { + const object = store.server(uid); + return { uid, name: object.metadata.name, kind: object.kind, data: object.data }; + }) + .sort((a, b) => a.name.localeCompare(b.name) || a.uid.localeCompare(b.uid)); + +const handle = connectNativeWatch( + url, + (event) => { + applyStreamEvent(store, event); + report({ event: event.type, members: members() }); + }, + { retryDelayMs: 100, onError: (code, message, terminal) => report({ error: { code, message, terminal } }) }, +); +handle.subscribe((state) => report({ state: state.status })); + +const input = createInterface({ input: process.stdin }); +input.on("line", (line) => { + if (line.trim() === "close") handle.close(); +}); +input.on("close", () => handle.close()); +await handle.closed.finally(() => input.close()); From 2399ed30896e6dff8663be68f5d81cdd8339773b Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 5 Oct 2026 11:27:25 +0000 Subject: [PATCH 5/6] docs: document native viewing and mark slice 1 implemented Add a native-proxy usage snippet and a source-choice table to the README, a native connections reference to the client README, and point the guides, proposals, spec note and examples at the implemented read-only connector. Record slice 1's acceptance evidence and list the follow-ups separately: native editing, native resume/streaming lists/pagination, comparative measurements, gateway continuation and save progress, and editor API cleanup. Co-Authored-By: Claude Opus 5.5 --- CONTRIBUTING.md | 12 ++-- README.md | 72 ++++++++++++++----- Taskfile.yml | 2 +- conformance/README.md | 5 +- docs/adopting.md | 9 +-- docs/alternatives.md | 2 +- docs/auth.md | 5 +- docs/client-state-model.md | 9 ++- docs/field-reports/third-our-identity.md | 32 +++++++-- docs/glossary.md | 2 +- docs/migrating.md | 4 +- ...006-stream-and-save-implementation-plan.md | 35 +++++---- .../0009-stream-and-editor-separation.md | 6 +- docs/proposals/0010-gateway-api-cleanup.md | 6 +- docs/saving.md | 4 +- docs/why-a-gateway.md | 40 ++++++----- examples/README.md | 12 ++-- examples/vanilla-browser/README.md | 13 ++-- gateway/README.md | 8 +-- packages/krm-stream/README.md | 65 +++++++++++++---- spec/v1.md | 9 +-- 21 files changed, 227 insertions(+), 125 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index f2e373b..9bceb6a 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -16,10 +16,10 @@ Run `task fixtures-check`, `task test`, and `task lint` before opening a pull re ## Design rules Start with watch streams and authoritative resource views; add editing as an optional layer. -The [README](README.md) explains current capabilities and the native-fetch direction. Current API -examples must match this branch. The next native slice is read-only LIST/WATCH using the existing -lifecycle and state-input helper; editor cleanup, native editing and larger comparisons are separate. -Native transport and save-progress evaluation remain proposals until implemented. +The [README](README.md) explains current capabilities and how to choose a native or gateway source. +Current API examples must match this branch. Native viewing is read-only LIST/WATCH on the shared +connection lifecycle; native editing, native resume, editor cleanup and larger comparisons are +separate. Save-progress evaluation remains a proposal until implemented. - Keep the core gateway free of `client-go`; Kubernetes integration belongs in `gateway/kube`. - Keep the browser client framework-free and free of runtime dependencies. @@ -62,9 +62,9 @@ the rule it protects. |---|---| | `task test` | Deterministic gateway, adapter, client, and shared-fixture coverage. | | `task e2e-wire` | Real Go SSE bytes consumed by the TypeScript client over HTTP. | -| `task e2e-browser` | Native `EventSource` and unbundled ESM in Chromium. | +| `task e2e-browser` | Native `EventSource`, unbundled ESM and the native viewer example in Chromium. | | `task cluster-facts` | Record observed Kubernetes behavior for the supported cluster version. | -| `task test-real-api` | Compose the save path against a real API server: gateway, host endpoint and browser store. | +| `task test-real-api` | Compose the save path and native viewing against a real API server: gateway or host proxy, host endpoint and browser store. | | `task test-cluster` | Exercise the Kubernetes backend against a real API server. | The cluster tasks need Docker and take longer; the fixture suites are the per-pull-request baseline. diff --git a/README.md b/README.md index 23108f3..cc59785 100644 --- a/README.md +++ b/README.md @@ -13,9 +13,11 @@ Efficient live Kubernetes views for browser applications, with optional editing. `krm-stream` helps applications watch Kubernetes resources while controllers and other users keep -changing them. Its gateway delivers a defined resource view, withholds selected values, suppresses -irrelevant updates and optionally shares upstream watches. Its headless TypeScript client adds connection lifecycle, recovery and -live state and reconciles incoming changes with local drafts when a page needs editing. +changing them. Its headless TypeScript client adds connection lifecycle, recovery and a live resource +store, reading native Kubernetes collections through a host proxy or projected views from the +gateway, and reconciles incoming changes with local drafts when a page needs editing. Its gateway +delivers a defined resource view, withholds selected values, suppresses irrelevant updates and +optionally shares upstream watches. Your application supplies authentication, authorization policy, Kubernetes credentials, UI and writes. The browser client has zero runtime dependencies and chooses no UI framework. @@ -51,18 +53,52 @@ This is a live state feed: intermediate updates may be coalesced. See [watching Start with the resource state and guarantees the page needs; framing is an implementation detail. -| Source | Use it for | Current support | +| Source | What the page receives | Connector | |---|---|---| -| Native through a host proxy | Original Kubernetes resources without adopting the gateway; credentials stay on the host | A focused read-only fetch connector is [next work](docs/field-reports/third-our-identity.md#native-watch-connector) | -| Gateway | Named views, Secret-value redaction, suppression and optional upstream sharing | Supported today through the fetch/SSE connector | +| Native through a host proxy | Original authorized resources, with the shared client lifecycle and store | `connectNativeWatch`, read-only: LIST, then WATCH, and a fresh LIST on every reconnect | +| Gateway | Projected, redacted or suppressed views and optional upstream watch sharing | `connectResourceStream`, delivered over SSE | -Native is the straightforward starting point for hosts that already proxy Kubernetes. Gateway SSE -is the delivery format for projected views; its added capabilities remain useful. Both sources should -reuse connection state, errors, cancellation, bounded recovery and state application. -The first native slice lists then watches and re-lists on reconnect. Native editing, resume and a -larger comparison example follow separately. No source fallback may bypass a refused view. +Native is the straightforward starting point for hosts that already proxy Kubernetes. It delivers +what the proxy returns, Secret values and machinery fields included; it provides no projection, +redaction, suppression or watch sharing. Both connectors share connection state, errors, +cancellation, bounded recovery and state events. Use a separate store per source, scope and login +identity, and never fall back from a refused gateway view to native access. -## Watch a resource view today +## Watch native resources through a host proxy + +```ts +import { + LiveResourceStore, readOnlyPolicy, applyStreamEvent, + connectNativeWatch, nativeCollectionURL, +} from "@configbutler/krm-stream"; + +const store = new LiveResourceStore(readOnlyPolicy); +const stopRendering = store.subscribe(() => renderResources(store)); +const connection = connectNativeWatch( + // → /k8s/apis/apps/v1/namespaces/app/deployments?labelSelector=tier%3Dweb + nativeCollectionURL("/k8s", { + group: "apps", version: "v1", resource: "deployments", namespace: "app", labelSelector: "tier=web", + }), + event => applyStreamEvent(store, event), +); +const stopConnection = connection.subscribe(state => renderConnection(state.status)); +connection.closed.catch(reportApplicationError); + +// On view disposal: +stopRendering(); +stopConnection(); +connection.close(); +``` + +The host mounts the API server's paths under a base such as `/k8s` and owns credentials, routing and +which collections a user may read. The connector lists the complete collection, delivers it as a +snapshot and becomes `live` once the WATCH from the collection's resourceVersion is accepted. Every +reconnect lists again, and the store prunes only when that snapshot completes. HTTP or in-stream 410 +recovers within the bounded retry budget; 401, 403 and other 4xx are terminal. Slice 1 does not +paginate: a LIST that returns a continuation token is refused. The +[native viewer example](examples/native-viewer/README.md) runs this against `kubectl proxy`. + +## Watch a gateway view ```ts import { @@ -88,7 +124,7 @@ stopConnection(); connection.close(); ``` -The current viewer uses `LiveResourceStore(readOnlyPolicy)`; a dedicated read-only store is deferred. +Viewers use `LiveResourceStore(readOnlyPolicy)`; a dedicated read-only store is deferred. The connector delivers state events independently of editing. It uses same-origin cookies by default, exposes connection state and bounded retries, and stops on terminal refusals. Apply each event synchronously. The [client README](packages/krm-stream/README.md) explains lifecycle and errors. @@ -134,11 +170,14 @@ arrays, and [saving edits safely](docs/saving.md) for the complete host-owned wr ```mermaid flowchart LR api["Kubernetes API"] + proxy["Your native proxy
Credentials and routing"] gateway["Go gateway
Scopes, views and optional sharing"] connector["Fetch connector
State events and recovery"] store["Resource store
Live state and optional drafts"] ui["Your list, viewer or form"] save["Your save endpoint
Authorize, validate and conditionally PATCH"] + api -->|"LIST and WATCH"| proxy + proxy -->|"Native watch JSON"| connector api -->|"Snapshot and watch"| gateway gateway -->|"SSE"| connector connector --> store @@ -148,7 +187,8 @@ flowchart LR save --> api ``` -The gateway runs inside your Go application. Per-user backends let Kubernetes authorize the caller's +A page uses one source per store: the native proxy or the gateway. The gateway runs inside your +Go application. Per-user backends let Kubernetes authorize the caller's reads; a shared backend uses a service identity and requires checks for each subscriber. Sharing reduces duplicate upstream work; access controls and host limits govern who can consume it. Gateway upstream continuation and improved save progress during suppressed churn are proposed work. @@ -168,7 +208,7 @@ such as `spec`, `status` or ConfigMap `data`. Custom resources follow the same c |---|---| | `github.com/ConfigButler/krm-stream/gateway` | Dependency-free Go stream gateway and SSE handler | | `github.com/ConfigButler/krm-stream/gateway/kube` | Optional client-go backend and SubjectAccessReview authorizer | -| `@configbutler/krm-stream` | Dependency-free ESM connector and resource/editor store | +| `@configbutler/krm-stream` | Dependency-free ESM native and gateway connectors, and the resource/editor store | | [spec/v1.md](spec/v1.md) and [conformance](conformance/README.md) | Shared normative contract and executable fixtures | ## Guides @@ -178,7 +218,7 @@ such as `spec`, `status` or ConfigMap `data`. Custom resources follow the same c - [Editor state model](docs/client-state-model.md): drafts, conflicts, redactions and arrays. - [Saving](docs/saving.md): conditional writes, recovery and user-facing outcomes. - [Authorization](docs/auth.md) and [operations](docs/operations.md): identity, revocation and runtime limits. -- [Examples](examples/README.md): browser, conditional editor, recovery recipes and Vue integration. +- [Examples](examples/README.md): native viewer, gateway browser demo, conditional editor, recovery recipes and Vue integration. - [Delivery plan](docs/proposals/0006-stream-and-save-implementation-plan.md): completed work, open work, ordering and dependencies. - [Upgrading from 0.7](docs/migrating.md) and [releasing](docs/releasing.md). diff --git a/Taskfile.yml b/Taskfile.yml index 29584a0..f1a89bc 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -440,7 +440,7 @@ tasks: # The save path, composed against a real API server: the real gateway, a real host endpoint and the # real browser store (run by node). test-real-api: - desc: "Real-API save composition: status churn, overlapping recovery and the UID race (proposal 0006)." + desc: "Real-API composition: save path (status churn, overlapping recovery, UID race) and native viewing through a host proxy." deps: [cluster-up] dir: gateway/kube env: diff --git a/conformance/README.md b/conformance/README.md index 1b01812..6c5dd65 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -1,8 +1,9 @@ # conformance — the shared contract, executable These fixtures verify the current gateway SSE contract and browser state application. Editing cases -add local drafts to that live-state contract; viewing does not require edit controls. Requested native -watch support will need recovery tests of its own, not an assumption that these SSE fixtures prove it. +add local drafts to that live-state contract; viewing does not require edit controls. Native watch +support has recovery tests of its own (`packages/krm-stream/test/native.test.ts`); these SSE fixtures +do not prove it. **One YAML file describes one scenario end to end**: the Kubernetes watch input, the gateway's wire output, and the client's resulting state after applying events and local edits. The Go and TypeScript diff --git a/docs/adopting.md b/docs/adopting.md index bdfe4fd..0234d44 100644 --- a/docs/adopting.md +++ b/docs/adopting.md @@ -7,11 +7,12 @@ and browser reconciliation; your application owns identity, authorization policy Choose scope, projection and sharing separately using [watching resources](why-a-gateway.md). Choose the source for the guarantees the page needs: native access retains original resources through an existing host proxy; the gateway adds projected views, redaction, suppression and optional sharing, -delivered over SSE. Both should reuse a fetch-based frontend lifecycle. +delivered over SSE. Both connectors share one fetch-based frontend lifecycle. -The [first native slice](field-reports/third-our-identity.md#slice-1-a-native-viewer) is a small read-only -LIST/WATCH connector with re-list recovery and a viewer. Native editing is separate and does not block -that slice. Today the supported connector uses the gateway wiring below. +For native access, a host that already proxies Kubernetes needs no gateway: point +`connectNativeWatch` at a collection behind that proxy, as in the [README](../README.md#watch-native-resources-through-a-host-proxy) +and the [native viewer example](../examples/native-viewer/README.md). It is read-only and re-lists on +every reconnect; native editing is separate. The steps below wire the gateway source. ## 1. Mount the stream endpoint diff --git a/docs/alternatives.md b/docs/alternatives.md index 21885db..917501e 100644 --- a/docs/alternatives.md +++ b/docs/alternatives.md @@ -6,7 +6,7 @@ and optional draft reconciliation. Choose tools according to the part your appli | Need | Relevant approach | Where krm-stream fits | |---|---|---| -| Native watch through an existing browser proxy | Host proxy plus a native fetch/watch consumer | A read-only native connector with shared lifecycle is next work; gateway SSE delivers the supported projected source today. | +| Native watch through an existing browser proxy | Host proxy plus a native fetch/watch consumer | `connectNativeWatch` lists and watches through the proxy with the shared lifecycle and store, read-only; the gateway adds projected views over SSE. | | Kubernetes API access from a server | Kubernetes client libraries and raw watches | The Go adapter uses `client-go`; the gateway adds projections and browser snapshot framing. | | A complete Kubernetes UI | Dashboard applications and their plugin APIs | krm-stream supplies state and transport; the host builds the UI. | | Review and deliver configuration packages | KRM package and GitOps systems | The host can record accepted writes in its delivery workflow. | diff --git a/docs/auth.md b/docs/auth.md index 6ead23e..13f684a 100644 --- a/docs/auth.md +++ b/docs/auth.md @@ -12,8 +12,9 @@ Prefer a per-user backend: Kubernetes RBAC is then the boundary by construction. is in [adopting](adopting.md). Watch sharing is an efficiency choice; it never grants a subscriber the service identity's access. Full/spec Secret redaction also does not replace scope authorization. -These settings describe the current gateway. The requested native fetch connector uses a host proxy -and the same session boundary. Native retains original authorized resources; gateway SSE delivers +These settings describe the current gateway. The native connector reads through a host proxy behind +the same session boundary; the proxy, not the browser, holds Kubernetes credentials and authorizes +each collection. Native retains original authorized resources; gateway SSE delivers projected views. Neither source may bypass refusal or extend session lifetime. The first native slice is read-only and must stop on terminal authentication failure, with a new connection after access is restored; native editing is separate. diff --git a/docs/client-state-model.md b/docs/client-state-model.md index 6a7ed86..db5b381 100644 --- a/docs/client-state-model.md +++ b/docs/client-state-model.md @@ -19,12 +19,11 @@ The connector consumes and delivers state events independently of the editor. Do not mutate a `draft()` return value directly: reads are detached copies, and that bypasses edit policy and notifications. Independent sources/scopes need separate stores. Switching identity or view -must not silently reuse drafts, redactions or snapshot state. Native watch integration is -[requested](field-reports/third-our-identity.md#native-watch-connector). The first native slice uses the -store read-only; this guide's editing and save examples currently use gateway views. Native editing -requires a source-bound policy and host read/write contract in a later slice. +must not silently reuse drafts, redactions or snapshot state. The native connector, +`connectNativeWatch`, uses the store read-only; this guide's editing and save examples use gateway +views. Native editing requires a source-bound policy and host read/write contract in a later slice. -Keep the existing standalone `applyStreamEvent(store, event)` for the native viewer. A bound store +The native viewer uses the existing standalone `applyStreamEvent(store, event)`. A bound store method is proposed editor cleanup in proposal 0009, not a current API or a prerequisite for native transport. No new store is needed. diff --git a/docs/field-reports/third-our-identity.md b/docs/field-reports/third-our-identity.md index 0dd9cb9..4dfa3ee 100644 --- a/docs/field-reports/third-our-identity.md +++ b/docs/field-reports/third-our-identity.md @@ -1,7 +1,8 @@ # Watch streams, optional editing and native access -**Current request, 2026-10-05. Slice 1 is not implemented.** The implementation starting point includes -connector separation, Go API cleanup and real-API save tests. Use [proposal 0006](../proposals/0006-stream-and-save-implementation-plan.md#open-work-and-delivery-order) +**Request of 2026-10-05. Slice 1, the native viewer, is implemented; the follow-ups below remain +open.** It builds on the connector separation, Go API cleanup and real-API save tests. Use +[proposal 0006](../proposals/0006-stream-and-save-implementation-plan.md#open-work-and-delivery-order) for completed work, delivery order and independent tracks. ## Identity and ownership @@ -13,13 +14,13 @@ conflict review and captured save intent; the host owns every write. | Source | Delivered content | Added value | |---|---|---| -| Native through a host proxy — requested | Original Kubernetes resources the host authorizes, including Secret values and machinery fields | Reuse connection lifecycle and live state without adopting the gateway | -| Gateway — supported today | Named projected views delivered over SSE | Selected disclosure, fewer downstream events and optional shared upstream watches | +| Native through a host proxy — read-only viewing implemented | Original Kubernetes resources the host authorizes, including Secret values and machinery fields | Reuse connection lifecycle and live state without adopting the gateway | +| Gateway — supported | Named projected views delivered over SSE | Selected disclosure, fewer downstream events and optional shared upstream watches | Native is the straightforward entry point for a host that already proxies Kubernetes. Gateway SSE is the delivery format of the projected source, not a legacy label for its capabilities. Both sources -use fetch and should share frontend lifecycle and state application. SSE and native JSON can both -carry errors. The existing SSE connector already handles HTTP refusals and in-stream errors. +use fetch and share frontend lifecycle and state application. SSE and native JSON can both carry +errors; each connector classifies its own HTTP refusals and in-stream errors. Scope, projection, suppression and sharing have separate costs and guarantees; see [watching resources](../why-a-gateway.md). Full/spec redaction withholds core Secret values while @@ -34,6 +35,11 @@ wrapping native watch frames in SSE. Keep the projected gateway and its current ### Slice 1: a native viewer +**Implemented** as `connectNativeWatch` and `nativeCollectionURL`, with a shared internal lifecycle +used by both connectors, the [native viewer example](../../examples/native-viewer/README.md) and the +[client reference](../../packages/krm-stream/README.md#native-connections). The requirements below are +the contract it was built and tested against. + Deliver one small read-only connector, using the existing `ResourceStateEvent` consumer and `LiveResourceStore(readOnlyPolicy)`. Keep the standalone `applyStreamEvent` API for this slice; proposal 0009's bound-method/editor cleanup is independent. No second store, framework or native @@ -68,6 +74,12 @@ change; native editing and the larger comparison project do not gate it. ### Acceptance for slice 1 +**Met.** Deterministic fake-fetch tests in `packages/krm-stream/test/native.test.ts` cover the list +below; the existing SSE lifecycle suite passes unchanged. `task e2e-browser` loads the example on both +entry points against a Playwright-played proxy, and `TestRealAPINativeWatchThroughHostProxy` +(`task test-real-api`) runs the connector through a credential-holding `/k8s` proxy against a real +API server: typed-list type metadata, modification, deletion, selector exit and same-name recreation. + Tests cover an empty collection; complete-list validation, unexpected continuation and type metadata; chunked frames; malformed/truncated input; initial/list/watch failures; HTTP/in-stream 410 and auth refusal; transient retries, exhaustion and retry hints; close during fetch/read/backoff; host callback @@ -81,6 +93,14 @@ a new cluster campaign is not a prerequisite. Existing tests still apply to any ## Follow-ups after the viewer +Each is separate from slice 1 and from the others: + +- Native editing with machinery/last-applied protections and a source-bound write contract. +- Native resume, streaming lists and pagination. +- Comparative view and sharing measurements. +- Independent gateway continuation and save-progress evaluation. +- The independently proposed editor API cleanup in proposal 0009. + Native editing is a separate slice: define the editable policy, exclude machinery and last-applied annotation changes, and bind host reads/writes to the same source and UID. Preserve later typing, submitted intent and reconciliation guards. A native response must not recover a projected editor. diff --git a/docs/glossary.md b/docs/glossary.md index 7755909..a5b7140 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -57,7 +57,7 @@ and `synced`; only completion prunes unseen UIDs. A partial snapshot cannot esta **SSE (Server-Sent Events)** is the current gateway's text-event framing, consumed by the official connector through **fetch**. Browser `EventSource` is another SSE client with different header/retry limitations. Fetch can read SSE or native Kubernetes watch JSON and inspect HTTP responses; both -formats can carry errors. A native fetch/watch connector is requested, while SSE remains supported. +formats can carry errors. `connectNativeWatch` reads native watch JSON; `connectResourceStream` reads gateway SSE. **Gateway** is the embeddable Go read layer. It enforces host scope/view policy, projects resources and delivers the current SSE protocol. A host can separately proxy native Kubernetes access with a diff --git a/docs/migrating.md b/docs/migrating.md index 80dfef3..15725ba 100644 --- a/docs/migrating.md +++ b/docs/migrating.md @@ -4,8 +4,8 @@ The release after 0.7 has one browser connector and one name for each gateway se 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 on this branch. Proposed editor removals in proposal 0009 are not migration requirements -yet. A read-only native fetch/watch connector is separate requested work and uses today's standalone -state-input helper. This upgrade still uses gateway SSE; native viewing does not wait for editor cleanup. +yet. The read-only native connector, `connectNativeWatch`, is a new addition rather than a migration +step, and uses today's standalone state-input helper. Gateway pages keep `connectResourceStream`. For the overall direction, see [watching resources](why-a-gateway.md). ## Browser client diff --git a/docs/proposals/0006-stream-and-save-implementation-plan.md b/docs/proposals/0006-stream-and-save-implementation-plan.md index b79240f..3a625cd 100644 --- a/docs/proposals/0006-stream-and-save-implementation-plan.md +++ b/docs/proposals/0006-stream-and-save-implementation-plan.md @@ -1,13 +1,14 @@ # Proposal 0006: Delivery plan for watch streams and optional editing **Status: active delivery plan, updated 2026-10-05.** This is the single inventory of completed work, -open work and ordering. The starting implementation includes connector separation, Go API cleanup -and real-API save-composition tests. Detailed contracts stay in the linked proposals and guides +open work and ordering. The implementation includes connector separation, Go API cleanup, +real-API save-composition tests and the read-only native viewer. Detailed contracts stay in the linked proposals and guides rather than becoming another roadmap. Start with [watching resources](../why-a-gateway.md); add the [editor](../client-state-model.md) only -where a page needs drafts. Prioritize a read-only native fetch connector for hosts with a proxy; -the gateway remains the projected source with redaction, suppression and sharing, delivered over SSE. +where a page needs drafts. Hosts with a Kubernetes proxy can read native collections with the +read-only native connector; the gateway remains the projected source with redaction, suppression and +sharing, delivered over SSE. Follow the [design rules](../../CONTRIBUTING.md#design-rules) and [release policy](../releasing.md). [Proposal 0005](0005-kubernetes-stream-and-save-semantics.md) owns the stream/save tradeoffs. @@ -28,6 +29,7 @@ remain separate delivery steps. | Optional editor and safe-save baseline | Draft reconciliation, explicit conflicts, atomic intent capture and guarded projected reads; recovery remains review plus another deliberate Save | [Editor model](../client-state-model.md), [saving](../saving.md) | | Editor integration and recovery recipes | Tested deletion recovery, keep-local resolution and Vue subscription ownership | [Recipes](../../examples/editor-recipes/README.md), [Vue](../../examples/vue/README.md) | | Real-API save composition | Cases for suppressed churn, guarded-read overlap, structured errors and same-name UID replacement | [Completed baseline](#completed-baseline) | +| Native viewer (slice 1) | `connectNativeWatch` and `nativeCollectionURL`: unpaginated LIST, WATCH from the collection RV, fresh LIST on reconnect, native 410 recovery and terminal refusals on a lifecycle shared with the gateway connector; fake-fetch, Chromium and real-API tests; minimal example | [Request](../field-reports/third-our-identity.md#slice-1-a-native-viewer), [example](../../examples/native-viewer/README.md), [client](../../packages/krm-stream/README.md#native-connections) | | Watch-first documentation | README, watch/edit/save guides, native-fetch request and compact decision records; duplicate/superseded guides removed | [README](../../README.md), [current request](../field-reports/third-our-identity.md) | ## Open work and delivery order @@ -37,7 +39,7 @@ before starting a later one. Deliver separately reviewable changes and use the p | Order | Open work | Dependencies and completion | |---|---|---| -| 1 | Native viewer: a short contract, shared lifecycle refactor, LIST/WATCH connector, tests and minimal example in one focused PR | Use today's event consumer and read-only store; re-list on reconnect. Native editing, editor cleanup, resume and benchmarks do not gate this slice. Acceptance lives in the [request](../field-reports/third-our-identity.md#slice-1-a-native-viewer). | +| 1 | Native viewer — **completed** | See [completed work](#completed-work); acceptance evidence is recorded in the [request](../field-reports/third-our-identity.md#acceptance-for-slice-1). | | 2 | Native editing | After the viewer; define source-bound edit policy and host validation, protect machinery fields and preserve save guards. Coordinate any public editor changes with proposal 0009. | | 3 | Native reconnect efficiency | After the viewer; evaluate resumable watches, streaming-list initialization and pagination with explicit membership/checkpoint tests. Do not claim these efficiencies for slice 1. | | 4 | Gateway upstream continuation | Independent track with a design below; review reopen/credential bounds, record a baseline, implement and measure. No dependency on native transport. | @@ -48,22 +50,17 @@ before starting a later one. Deliver separately reviewable changes and use the p ### Next implementation -Start the native viewer from the branch containing the implemented connector separation and Go API -cleanup. Keep native frame parsing separate from SSE while sharing lifecycle and useful HTTP helpers. -The short read-only contract and test results belong in the implementation PR; no additional -editor-design review or comparison campaign must finish first. +The native viewer is in place: native frame parsing stays separate from SSE, and both connectors share +the lifecycle and HTTP helpers. It re-lists on every reconnect and refuses paginated responses; it +claims no resume efficiency. Choose the next change from orders 2 and 3, or from the independent +tracks; none of them waits for another. -Preserve existing SSE behavior and public APIs during the lifecycle refactor. Slice 1 requires complete -LIST membership, correct type/UID handling, an accepted WATCH before live readiness, transport-specific -410 recovery, terminal refusals, bounded repeated failure and cancellation. Use unpaginated lists; -reject unexpected continuation rather than pruning a partial snapshot. Add one small runnable native -viewer and document the two sources. See the [request](../field-reports/third-our-identity.md#acceptance-for-slice-1). - -Keep the current standalone state-input helper. Proposed editor removals remain in +Native editing (order 2) builds on the viewer without waiting for resume or streaming lists. Native +reconnect efficiency (order 3) needs its own membership and checkpoint tests. Keep the current +standalone state-input helper; proposed editor removals remain in [proposal 0009](0009-stream-and-editor-separation.md#proposed-editor-cleanup), with brief API change notes -when implemented. Native editing can follow the viewer without waiting for resume or streaming lists. -Gateway continuation and save measurements can proceed independently. Retain the existing projected -save flow and explicit protocol-version policy; neither needs redesign to deliver a native viewer. +when implemented. Gateway continuation and save measurements proceed independently. Retain the existing +projected save flow and explicit protocol-version policy. ## Deferred work diff --git a/docs/proposals/0009-stream-and-editor-separation.md b/docs/proposals/0009-stream-and-editor-separation.md index e3fc5d5..d605ad9 100644 --- a/docs/proposals/0009-stream-and-editor-separation.md +++ b/docs/proposals/0009-stream-and-editor-separation.md @@ -24,9 +24,9 @@ rather than duplicating them in this implementation plan. The [native-watch request](../field-reports/third-our-identity.md#native-watch-connector) extends this separation: native is the original-resource source through a host proxy; the gateway provides projected views over SSE. Share lifecycle and state application without converting native JSON to SSE. -The first native slice is a read-only viewer using today's event consumer and standalone -`applyStreamEvent`. It does not depend on the editor cleanup below or a new store. Native support -is not implemented here; native editing follows separately. +The first native slice, `connectNativeWatch`, is a read-only viewer using today's event consumer and +standalone `applyStreamEvent`. It did not depend on the editor cleanup below or a new store; native +editing follows separately. ## Implemented connector and correctness fixes diff --git a/docs/proposals/0010-gateway-api-cleanup.md b/docs/proposals/0010-gateway-api-cleanup.md index 140653b..ef98b3c 100644 --- a/docs/proposals/0010-gateway-api-cleanup.md +++ b/docs/proposals/0010-gateway-api-cleanup.md @@ -9,9 +9,9 @@ PR #54. Use [the gateway README](../../gateway/README.md) for current setup and Keep one name per API, share stream configuration and remove repository-only runtime exports before 1.0. Preserve authorization, named projections, bounded HTTP delivery, shared-watch lifecycle and v1 wire behavior. The gateway remains the projected watch layer and current SSE delivery path; -[native browser transport](../field-reports/third-our-identity.md#native-watch-connector) is separate -requested work. The gateway remains the projected source over SSE; a native viewer does not require -changes to its APIs or framing. +[native browser transport](../field-reports/third-our-identity.md#native-watch-connector) is separate. +The gateway remains the projected source over SSE; the native viewer required no change to its APIs +or framing. | Area | Implemented result | |---|---| diff --git a/docs/saving.md b/docs/saving.md index e278d9d..697236a 100644 --- a/docs/saving.md +++ b/docs/saving.md @@ -18,8 +18,8 @@ accepted write and application progress are separate states; neither a clean for proves a rollout or Git workflow completed. A definite 409 rejection differs from a lost response whose write outcome is unknown; do not blindly replay an ambiguous write. -This guide describes projected editing through the current gateway. The requested native connector -starts with read-only viewing. Native editing is a later slice requiring an explicit read/write and +This guide describes projected editing through the current gateway. The native connector is +read-only. Native editing is a later slice requiring an explicit read/write and edit-policy contract before sharing these recipes. A raw response must never be fed into a projected editor as recovery or save confirmation. diff --git a/docs/why-a-gateway.md b/docs/why-a-gateway.md index 1e69d52..27e38f9 100644 --- a/docs/why-a-gateway.md +++ b/docs/why-a-gateway.md @@ -60,9 +60,9 @@ recreated resource from an earlier object with the same name. The fetch connector exposes `connecting`, `syncing`, `live`, `retrying`, `closed`, `terminal` and `exhausted`. Apply state events synchronously before rendering completion. Retryable failures recover -within the connector's bounded budget. For the current SSE connector, every HTTP 4xx except 408/429, -terminal stream errors and a mismatched protocol header stop the connection. Native history-expiry -410 needs different recovery; it must be classified by native transport, not this SSE rule. +within the connector's bounded budget. For the gateway SSE connector, every HTTP 4xx except 408/429, +terminal stream errors and a mismatched protocol header stop the connection. The native connector +classifies HTTP and in-stream 410 (expired watch history) as recoverable by a fresh LIST instead. Dispose subscriptions and close the handle when their owner leaves. See the [client lifecycle reference](../packages/krm-stream/README.md#connections). @@ -86,23 +86,25 @@ Sharing reduces duplicate resource consumption; it does not supply a rate limite ## Native and gateway sources Native access is the straightforward starting point for a host that already proxies Kubernetes. -The requested native fetch connector adds lifecycle and state-store reuse while retaining original -resource fields. The gateway adds defined views, redaction, suppression and watch sharing, delivered +The native connector, `connectNativeWatch`, adds lifecycle and state-store reuse while retaining +original resource fields. The gateway adds defined views, redaction, suppression and watch sharing, delivered over SSE. Choose the guarantees the page needs. Native access alone does not provide those gateway capabilities, and `krm-raw/v1` still differs from native data. Both sources use fetch; native Kubernetes JSON and gateway SSE are different framing formats, and -both can carry errors. The existing SSE connector already handles refusal, expiry, cancellation and -bounded retries. Browser `EventSource` is a different client mechanism with header/retry limitations. - -The [first native slice](field-reports/third-our-identity.md#slice-1-a-native-viewer) is a read-only -viewer: ordinary LIST, then WATCH from the collection version, with a fresh LIST on every reconnect. -It reuses the existing lifecycle and `LiveResourceStore(readOnlyPolicy)`. HTTP/in-stream 410 recovers -within the bounded policy; terminal auth refusal never selects another source. An unexpected paginated -response must be refused rather than marked complete. Native editing, resume and streaming lists follow -separately; neither editor cleanup nor a comparative benchmark blocks the viewer. - -Today use the [gateway wiring](adopting.md); native code snippets will be added with the implementation. -A same-origin, cookie-authenticated host proxy keeps Kubernetes credentials server-side. Keep stores -separate across sources, scopes and identities and never fall back from a refused projected source -into native access. The native connector is not implemented by this documentation change. +both can carry errors. Both connectors handle refusal, expiry, cancellation and bounded retries +through one shared lifecycle; each classifies its own framing and HTTP responses. Browser `EventSource` is a different client mechanism with header/retry limitations. + +The native connector is a read-only viewer source: an ordinary LIST, then a WATCH from the +collection's resourceVersion, with a fresh LIST on every reconnect. It uses the shared lifecycle and +`LiveResourceStore(readOnlyPolicy)`, and is `live` only once the snapshot is applied and the WATCH is +accepted. HTTP/in-stream 410 recovers within the bounded policy; terminal auth refusal never selects +another source. A paginated LIST response is refused rather than marked complete. Native editing, +resume, streaming lists and pagination follow separately. + +See the [native usage snippet](../README.md#watch-native-resources-through-a-host-proxy), the +[client reference](../packages/krm-stream/README.md#native-connections) and the +[native viewer example](../examples/native-viewer/README.md); use the [gateway wiring](adopting.md) for +projected views. A same-origin, cookie-authenticated host proxy keeps Kubernetes credentials +server-side. Keep stores separate across sources, scopes and identities and never fall back from a +refused projected source into native access. diff --git a/examples/README.md b/examples/README.md index c2ba10a..5542a45 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,11 +1,13 @@ # Examples Start with [watching resources](../docs/why-a-gateway.md) and supported host wiring, then add the -optional editor. The checked browser example in [`vanilla-browser/`](vanilla-browser/) runs against -the current gateway source over SSE. A minimal native viewer is -[next work](../docs/field-reports/third-our-identity.md#slice-1-a-native-viewer); it reuses the current -read-only store and lifecycle. Native editing and the larger view/sharing comparison follow separately. -For current host integration patterns, use these small recipes: +optional editor. + +[Native viewer](native-viewer/README.md) is a read-only page that lists and watches native resources +through an existing host proxy, such as `kubectl proxy`, with the shared lifecycle and read-only store. +The checked browser example in [`vanilla-browser/`](vanilla-browser/) runs against the gateway source +over SSE. Native editing and the larger view/sharing comparison follow separately. +For gateway host integration patterns, use these small recipes: - [Same-origin cookie application](../docs/adopting.md#1-mount-the-stream-endpoint): fetch connector, session cookie, dynamic client acting as the user. diff --git a/examples/vanilla-browser/README.md b/examples/vanilla-browser/README.md index 25b4de6..73131fe 100644 --- a/examples/vanilla-browser/README.md +++ b/examples/vanilla-browser/README.md @@ -11,7 +11,8 @@ task demo Use `task e2e-browser` to run the Chromium check. It loads both published entry points, and also reads the gateway with a raw native `EventSource` to check its framing, reconnect snapshots and -terminal shutdown. +terminal shutdown. The same run loads the [native viewer](../native-viewer/README.md) against a +Playwright-played host proxy. The page demonstrates live status updates, three-way conflict handling, draft edits, and redacted Secret values. Useful fixtures include: @@ -27,9 +28,9 @@ Secret values. Useful fixtures include: Add `pace=0ms` for the fastest replay or use a positive value to inspect each event. -The example uses today's projected SSE connector over fetch. For a read-only page, render server -state with `LiveResourceStore(readOnlyPolicy)` as shown in the [README](../../README.md#watch-a-resource-view-today). +The example uses the gateway's projected SSE connector over fetch. For a read-only page, render server +state with `LiveResourceStore(readOnlyPolicy)` as shown in the [README](../../README.md#watch-a-gateway-view). Editing is an optional layer; the [editor model](../../docs/client-state-model.md) and -[saving guide](../../docs/saving.md) describe its intended use. A small native viewer is the next -slice; native editing and a larger source comparison follow separately. They are not features of -this gateway replay demo. +[saving guide](../../docs/saving.md) describe its intended use. For native resources through a host +proxy, see the [native viewer](../native-viewer/README.md), which this suite also loads; native +editing and a larger source comparison follow separately. diff --git a/gateway/README.md b/gateway/README.md index f0f3368..2c899a2 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -5,10 +5,10 @@ to Kubernetes-style `Backend` watches. It delivers live state for viewers and op It is transport-neutral until `Handler` or `ServeStream` adds the current SSE framing. Core has no Kubernetes client dependency; [`gateway/kube`](kube/) provides the client-go backend and authorizer. -A fetch-based native browser connector is [requested work](../docs/field-reports/third-our-identity.md#native-watch-connector). -Native is the original-resource source; this gateway delivers projected views over SSE. Native -access alone does not provide these projected-view guarantees. A small native viewer can ship -independently of native editing or changes to this gateway. See [watching resources](../docs/why-a-gateway.md) for the overall contract. +The browser client can also read native collections through a host's Kubernetes proxy with +`connectNativeWatch`, without this gateway. Native is the original-resource source; this gateway +delivers projected views over SSE. Native access alone does not provide these projected-view +guarantees, and native viewing required no change to this gateway. See [watching resources](../docs/why-a-gateway.md) for the overall contract. ```go handler := gateway.Handler(gateway.Options{ diff --git a/packages/krm-stream/README.md b/packages/krm-stream/README.md index 4e4bcbb..7b61cca 100644 --- a/packages/krm-stream/README.md +++ b/packages/krm-stream/README.md @@ -1,22 +1,20 @@ # @configbutler/krm-stream -`@configbutler/krm-stream` is the official dependency-free ESM client for consuming a KRM resource -stream in a browser or JavaScript application. It provides: +`@configbutler/krm-stream` is the official dependency-free ESM client for live Kubernetes resource +state in a browser or JavaScript application, with optional editing. It provides: -- `connectResourceStream`, the connector: it reads the stream over fetch, with session cookies or - bearer headers, and hands each resource state event to a callback, with bounded recovery. +- `connectNativeWatch`, which reads a native Kubernetes collection through a host proxy with LIST + and WATCH, and `nativeCollectionURL` to address one. +- `connectResourceStream`, which reads a gateway's projected views over SSE, and + `resourceStreamURL` for the v1 scope query format. - `LiveResourceStore` for server state, local drafts, conflicts, redactions, and merge patches, and `applyStreamEvent` to apply a state event to it. -- `resourceStreamURL` for the v1 scope query format. -Start with a watch-backed list or viewer; add draft reconciliation only when the page needs editing. -The package is headless and does not choose a UI framework. Today it consumes gateway SSE through -fetch, from the Go gateway or another v1 producer. A fetch-based native Kubernetes watch connector -is [next work](../../docs/field-reports/third-our-identity.md#slice-1-a-native-viewer): a read-only -LIST/WATCH source with re-list recovery and a small viewer, reusing the same lifecycle and store. -Native retains original authorized resources; gateway SSE delivers projected views with redaction, -suppression and optional sharing. The native connector is not implemented yet; editing and resume -are later slices. Keep current `applyStreamEvent` unchanged for the viewer. +Both connectors use fetch, share one connection lifecycle and hand each resource state event to a +callback, with bounded recovery. Native access delivers original authorized resources; the gateway +adds projected views, redaction, suppression and optional upstream watch sharing. Start with a +watch-backed list or viewer; add draft reconciliation only when the page needs editing. The package +is headless and does not choose a UI framework. Native viewing is read-only in this release. ```ts import { LiveResourceStore, readOnlyPolicy, applyStreamEvent, connectResourceStream, resourceStreamURL } from "@configbutler/krm-stream"; @@ -30,6 +28,8 @@ connection.closed.catch(reportApplicationError); // Close connection and dispose rendering subscriptions on view teardown. ``` +See [native connections](#native-connections) for the same viewer over a host's Kubernetes proxy. + ## Vendoring it without a bundler The default entry point is the per-module build: `dist/index.js` imports `./store.js`, which imports @@ -75,7 +75,7 @@ unsubscribe(); connection.close(); ``` -`connectResourceStream` is the current gateway connector. It uses fetch for same-origin cookies or +`connectResourceStream` is the gateway connector. It uses fetch for same-origin cookies or 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`. @@ -119,6 +119,43 @@ There is no official `EventSource` helper. It cannot send explicit authorization reconnect timing to the browser; a host-owned helper must handle protocol sequence gaps and terminal errors itself. Gateway framing still supports browser `EventSource` with session cookies (spec §7). +### Native connections + +```ts +const connection = connectNativeWatch( + nativeCollectionURL("/k8s", { version: "v1", resource: "configmaps", namespace: "app", name: "settings" }), + event => applyStreamEvent(store, event), // a LiveResourceStore(readOnlyPolicy) for this source only + { onError: (code, message, terminal) => showStreamError(code, message, terminal) }, +); +``` + +`connectNativeWatch` takes the same options and returns the same handle, states and completion +guarantees as `connectResourceStream`. `nativeCollectionURL` builds core (`/api/v1`) and grouped +(`/apis//`) collection paths, namespaced or cluster-wide, with an optional +`labelSelector`; `name` becomes `fieldSelector=metadata.name=`, so a named object can be absent, +deleted and recreated like any member. The host mounts the API server under the base, such as `/k8s`, +and owns credentials, routing and authorization; the collection URL must not set `watch`, +`resourceVersion`, `limit` or `continue`, which the connector controls. + +Each connection performs an ordinary LIST, validates every item's UID and name, and fills a missing +item `apiVersion`/`kind` from a typed collection (`ConfigMapList` lists `ConfigMap`s; type metadata +already present is kept). It delivers `reset` and an `added` per member, then opens a WATCH from the +collection's `resourceVersion` with the same selectors, and delivers `synced` — publishing `live` — +only once that WATCH is accepted. Watch events map to `added`, `modified` and `deleted` (with the +identity of the deleted object); bookmarks change nothing. Every reconnect lists again, so prior +state is kept until a replacement snapshot completes and prunes what it no longer lists: deletes and +selector exits missed while disconnected, or a same-name object recreated with a new UID. + +HTTP or in-stream 410 (expired watch history), 408, 429, 5xx, network failures, EOF, and malformed or +truncated frames are reported where applicable and consume the bounded retry budget; repeated expiry +cannot re-list in a tight loop. A Kubernetes `Status` message reaches `onError`, and its +`retryAfterSeconds` or an HTTP `Retry-After` sets the least the next reconnect waits. HTTP or in-stream +401, 403 and other 4xx are terminal. This release does not paginate: a LIST with a continuation token +is refused with a terminal error before anything is applied, because pruning from one page would +remove the others. Native objects carry whatever the proxy returns, including Secret values and +machinery fields; there is no projection, redaction or `seq`. Never fall back to native access after a +gateway refuses a view. + ## Optional editing Use `new LiveResourceStore()` for an editor, or provide a policy that narrows the form's editable diff --git a/spec/v1.md b/spec/v1.md index 660cd5e..2659f5d 100644 --- a/spec/v1.md +++ b/spec/v1.md @@ -4,10 +4,11 @@ > [the Go gateway](../gateway/README.md) and [the browser client](../packages/krm-stream/README.md). > It defines live projected resource state; draft editing is an optional consumer behavior. > -> A read-only native fetch connector is [requested work](../docs/field-reports/third-our-identity.md#slice-1-a-native-viewer). -> This specification describes the gateway's projected source over SSE. Native Kubernetes JSON is a -> different source format and does not use the SSE sequence or protocol-header rules. No wire behavior -> or conformance requirement changes in this documentation revision. +> The browser client also reads native Kubernetes collections through a host proxy +> ([`connectNativeWatch`](../packages/krm-stream/README.md#native-connections)). This specification +> describes the gateway's projected source over SSE. Native Kubernetes JSON is a different source +> format and does not use the SSE sequence or protocol-header rules; it changes no wire behavior or +> conformance requirement here. A resource stream is a one-way feed, over Server-Sent Events, of the current-and-then-live state of a **scope** (one named object, or a collection of a kind in a namespace). The server absorbs every From 68c9869dfaf0942013ef6b0f99a78f95c7089878 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 5 Oct 2026 12:25:40 +0000 Subject: [PATCH 6/6] fix(client): trim the native proxy base without a backtracking regex nativeCollectionURL trimmed trailing slashes with /\/+$/, which CodeQL flags as polynomial ReDoS (js/polynomial-redos) on caller input: a long run of slashes not at the end retries from every slash. A loop trims in linear time. Co-Authored-By: Claude Opus 5.5 --- packages/krm-stream/src/url.ts | 10 +++++++++- packages/krm-stream/test/native.test.ts | 10 ++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/packages/krm-stream/src/url.ts b/packages/krm-stream/src/url.ts index 8765aff..b215a5b 100644 --- a/packages/krm-stream/src/url.ts +++ b/packages/krm-stream/src/url.ts @@ -78,7 +78,7 @@ export function nativeCollectionURL(proxyBase: string, scope: NativeScope): stri } return encodeURIComponent(value); }; - const path = [proxyBase.replace(/\/+$/, "")]; + const path = [withoutTrailingSlashes(proxyBase)]; if (scope.group) path.push("apis", segment("group", scope.group)); else path.push("api"); path.push(segment("version", scope.version)); @@ -91,3 +91,11 @@ export function nativeCollectionURL(proxyBase: string, scope: NativeScope): stri const query = q.toString(); return path.join("/") + (query ? `?${query}` : ""); } + +/** The base without its trailing slashes. A loop, not `/\/+$/`: that pattern retries from every + * slash in a long run that does not end the string, which is quadratic in the caller's input. */ +function withoutTrailingSlashes(base: string): string { + let end = base.length; + while (end > 0 && base[end - 1] === "/") end--; + return base.slice(0, end); +} diff --git a/packages/krm-stream/test/native.test.ts b/packages/krm-stream/test/native.test.ts index d30b513..8fb959c 100644 --- a/packages/krm-stream/test/native.test.ts +++ b/packages/krm-stream/test/native.test.ts @@ -925,3 +925,13 @@ test("nativeCollectionURL refuses segments that would change the path", () => { "/k8s/api/v1/namespaces/a%20b/pods", ); }); + +test("nativeCollectionURL trims any run of trailing slashes from the proxy base in linear time", () => { + assert.equal(nativeCollectionURL("/k8s///", { version: "v1", resource: "pods" }), "/k8s/api/v1/pods"); + assert.equal(nativeCollectionURL("///", { version: "v1", resource: "pods" }), "/api/v1/pods"); + // A base that is all slashes but for its last character backtracks quadratically under /\/+$/. + const base = `${"/".repeat(100_000)}x`; + const started = performance.now(); + assert.equal(nativeCollectionURL(base, { version: "v1", resource: "pods" }), `${base}/api/v1/pods`); + assert.ok(performance.now() - started < 200, "trimming the proxy base must not backtrack"); +});