From 6adc4e21ed7e278d499a956b2d17cd88a8adec0f Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Mon, 5 Oct 2026 14:13:53 +0000 Subject: [PATCH] feat(client): resume native watches from a consumed checkpoint connectNativeWatch no longer re-lists on every reconnect. After a complete snapshot (LIST applied, WATCH accepted, synced consumed) each handle keeps a private checkpoint: the resourceVersion of the last event its consumer applied, or of the last bookmark. An EOF, network failure, truncated final frame or retryable refusal (408/429/5xx, HTTP or in-stream) resumes the WATCH from it with the same URL and selectors: no LIST, no reset, no synced, and the state goes connecting -> live on acceptance, so the store keeps membership, drafts and conflicts and receives missed changes as events. HTTP/in-stream 410 (or reason Expired/Gone), malformed frames, an in-stream error without a code, an event without a resourceVersion and terminal refusals discard the checkpoint; the next connection is a fresh LIST/WATCH snapshot within the bounded retry budget. An initialization interrupted before synced leaves no checkpoint. Every WATCH requests allowWatchBookmarks=true, now a reserved query parameter. Resource versions stay opaque and are never compared. Tests: fake-fetch resume, bookmark, opaque-RV, truncation, malformed, interrupted-initialization, expiry, terminal, transient, cancellation, per-handle and draft/conflict cases; native editor refusal during expiry recovery; Chromium viewer/editor resume and in-stream 410; and TestRealAPINativeResumeThroughHostProxy, whose host proxy drops, holds and expires watches (the expiry is induced by rewriting resourceVersion to 1; the 410 is the real API server's in-stream answer). Co-Authored-By: Claude Opus 5.5 --- CONTRIBUTING.md | 5 +- README.md | 11 +- docs/adopting.md | 4 +- docs/field-reports/third-our-identity.md | 52 +- docs/glossary.md | 9 + ...006-stream-and-save-implementation-plan.md | 40 +- docs/saving.md | 13 +- docs/why-a-gateway.md | 19 +- examples/native-editor/README.md | 7 +- examples/native-viewer/README.md | 12 +- .../tests/native-editor.spec.ts | 90 ++- .../tests/native-viewer.spec.ts | 106 +++- gateway/kube/native_e2e_test.go | 412 ++++++++++-- packages/krm-stream/README.md | 62 +- packages/krm-stream/e2e/native-driver.ts | 11 +- packages/krm-stream/src/lifecycle.ts | 5 +- packages/krm-stream/src/native.ts | 260 +++++--- .../krm-stream/test/native-editing.test.ts | 86 ++- packages/krm-stream/test/native.test.ts | 589 +++++++++++++++++- 19 files changed, 1570 insertions(+), 223 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b6acc9a..81656c9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -18,8 +18,9 @@ Run `task fixtures-check`, `task test`, and `task lint` before opening a pull re Start with watch streams and authoritative resource views; add editing as an optional layer. 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 LIST/WATCH on the shared -connection lifecycle, and native editing writes conditional merge patches back through the same -proxy; native resume, editor cleanup and larger comparisons are separate. Save-progress evaluation remains a proposal until implemented. +connection lifecycle and resumes ordinary reconnects from a per-handle checkpoint, and native editing +writes conditional merge patches back through the same proxy; streaming lists, pagination, 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. diff --git a/README.md b/README.md index 87e3f10..3fc2681 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ Start with the resource state and guarantees the page needs; framing is an imple | Source | What the page receives | Connector | |---|---|---| -| Native through a host proxy | Original authorized resources, with the shared client lifecycle and store | `connectNativeWatch`: LIST, then WATCH, and a fresh LIST on every reconnect; edits are written back through the same proxy | +| Native through a host proxy | Original authorized resources, with the shared client lifecycle and store | `connectNativeWatch`: LIST, then WATCH; a reconnect resumes the WATCH, and only an expired history lists again; edits are written back through the same proxy | | 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. It delivers @@ -92,10 +92,11 @@ 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. The connector does not -paginate: a LIST that returns a continuation token is refused. The +snapshot and becomes `live` once the WATCH from the collection's resourceVersion is accepted. A +dropped or ended watch resumes from the last event applied, without a new snapshot, so drafts stay +and missed changes arrive as events. HTTP or in-stream 410 (expired history) lists again within the +bounded retry budget, and the store prunes only when that snapshot completes; 401, 403 and other 4xx +are terminal. The connector 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 diff --git a/docs/adopting.md b/docs/adopting.md index c1e0a36..c87e998 100644 --- a/docs/adopting.md +++ b/docs/adopting.md @@ -11,8 +11,8 @@ delivered over SSE. Both connectors share one fetch-based frontend lifecycle. 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 re-lists on every -reconnect; to edit through the same proxy, see [native editing](saving.md#native-editing-through-a-host-proxy). +and the [native viewer example](../examples/native-viewer/README.md). A reconnect resumes the watch, +and only an expired history lists again; to edit through the same proxy, see [native editing](saving.md#native-editing-through-a-host-proxy). The steps below wire the gateway source. ## 1. Mount the stream endpoint diff --git a/docs/field-reports/third-our-identity.md b/docs/field-reports/third-our-identity.md index 7148238..7eacc95 100644 --- a/docs/field-reports/third-our-identity.md +++ b/docs/field-reports/third-our-identity.md @@ -1,7 +1,7 @@ # Watch streams, optional editing and native access -**Request of 2026-10-05. Slice 1, the native viewer, and slice 2, native editing, are implemented; -the other follow-ups below remain open.** It builds on the connector separation, Go API cleanup and real-API save tests. Use +**Request of 2026-10-05. Slice 1, the native viewer, slice 2, native editing, and native watch +resumption are implemented; the other 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. @@ -154,16 +154,56 @@ saved and echoed on both entry points and rendered a real 422. The comparative f Each is separate from the slices above and from the others: -- Native resume, streaming lists and pagination. +- Native resume — **implemented**, see [below](#native-watch-resumption); streaming lists and + pagination remain open. - Comparative view and sharing measurements. - Independent gateway continuation and save-progress evaluation. - The independently proposed editor API cleanup in proposal 0009. -Resumable native watches, streaming-list initialization and pagination are later improvements with -their own recovery tests. Slice 1 improves adoption and lifecycle reuse; re-listing on every reconnect -does not claim resume efficiency. Gateway upstream continuation has a different checkpoint owner and +Slice 1 re-listed on every reconnect and claimed no resume efficiency; the requirements above are +the contract it was built against. Gateway upstream continuation has a different checkpoint owner and can proceed independently. Neither implies browser replay for gateway SSE v1. +### Native watch resumption + +**Implemented** in `connectNativeWatch`; the contract is in the +[client reference](../../packages/krm-stream/README.md#native-connections). + +- Each handle keeps a private checkpoint, armed only once a complete snapshot was applied and its + WATCH accepted, and advanced only after the consumer applied an event or the stream passed a + bookmark. It is never exposed, accepted or shared, so it stays bound to the handle's proxy, scope, + selectors and credentials. Resource versions are opaque and replaced in stream order. Bookmarks + (`allowWatchBookmarks=true`) move only the checkpoint, never an object's write version. +- EOF, network failures, a truncated final frame and retryable refusals (408, 429, 5xx, HTTP or in + the stream) resume the WATCH from the checkpoint: no LIST, `reset` or `synced`, so membership, + drafts and conflicts survive and missed updates, deletes, selector exits and same-name UID + replacements arrive as watch events. The state goes `connecting` → `live` once the resumed watch is + accepted. +- HTTP or in-stream 410, malformed frames, an error without a code and an event without a + resourceVersion discard the checkpoint; the next connection is a fresh, complete LIST/WATCH + snapshot, during which the state is not live and editors refuse to save. Terminal refusals, the + bounded retry budget, cancellation and callback-error cleanup are unchanged; every reconnect still + consumes the budget. + +**Acceptance.** `packages/krm-stream/test/native.test.ts` covers missed updates, deletes, selector +exits and UID replacement replayed on resume, bookmarks, opaque versions, split and truncated frames, +malformed input, interrupted initialization, HTTP and in-stream expiry, repeated expiry to +exhaustion, terminal and transient refusals on a resumed watch, cancellation, per-handle checkpoints +and drafts/conflicts across a resume and a re-list; `native-editing.test.ts` covers a save refused +during expiry recovery. `task e2e-browser` runs the viewer and editor pages through resume and an +in-stream 410 on both entry points. `TestRealAPINativeResumeThroughHostProxy` (`task test-real-api`, +v1.36.4+k3s1) drops the watch at a host proxy, mutates the cluster while the next WATCH is held and +converges with no second LIST; it then rewrites a WATCH's resourceVersion to `1` so the real API server +answers with its own in-stream 410, and recovers through a fresh snapshot, twice in a row. The expiry +is induced by the proxy; the 410 is the server's. + +**Limits.** A resumed watch is live on acceptance, so an expired one is briefly live until its +in-stream 410 arrives; a resume does not invalidate a guarded read as a snapshot does (an event for +the object does), as on any live watch; the checkpoint is in memory and a page reload lists again. +Streaming-list initialization and pagination are evaluated in +[proposal 0006](../proposals/0006-stream-and-save-implementation-plan.md#3-native-streaming-lists-and-pagination) +and not implemented. + ## Show the value of each path After the viewer works, compare native/full/spec and shared/unshared gateway runs using the same diff --git a/docs/glossary.md b/docs/glossary.md index c5cc90a..f93895f 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -34,6 +34,15 @@ See [quiet streams and saving](saving.md#why-a-quiet-stream-can-still-reject-a-s as bookmarks and errors. History can expire and initialization/recovery must establish a complete collection boundary. The current gateway handles these mechanics for its browser clients. +**Bookmark** is a watch event that carries only a collection resource version: "the stream has +reached here". It changes no object and is not an object's write version. + +**Checkpoint** is the collection resource version a watch can resume from: the last event the client +applied, or the last bookmark. `connectNativeWatch` keeps one per connection handle, privately, and +**resumes** a dropped WATCH from it instead of listing again; the server replays what followed. When +the server reports the history expired (410), the checkpoint is discarded and a fresh snapshot +recovers. Gateway browser connections do not resume: each starts a snapshot. + **Scope** selects a target, resource kind, namespace, optional name and allowed labels. The host still authorizes the request; a selector does not grant permission. diff --git a/docs/proposals/0006-stream-and-save-implementation-plan.md b/docs/proposals/0006-stream-and-save-implementation-plan.md index a0ce8d9..274d72b 100644 --- a/docs/proposals/0006-stream-and-save-implementation-plan.md +++ b/docs/proposals/0006-stream-and-save-implementation-plan.md @@ -2,7 +2,8 @@ **Status: active delivery plan, updated 2026-10-05.** This is the single inventory of completed work, open work and ordering. The implementation includes connector separation, Go API cleanup, -real-API save-composition tests, the native viewer and native editing with a browser page. Detailed +real-API save-composition tests, the native viewer, native editing with a browser page and native +watch resumption. 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 @@ -29,8 +30,9 @@ 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) | +| Native viewer (slice 1) | `connectNativeWatch` and `nativeCollectionURL`: unpaginated LIST, WATCH from the collection RV, fresh LIST on reconnect (superseded by resumption below), 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) | | Native editing (slice 2) | Machinery (`managedFields`, last-applied annotation) read-only in the store under every policy; maps holding a protected path merged key by key; `nativeObjectURL`; `gateway.ValidateNativeMergePatch` for host proxies; a native conditional editor with guarded native recovery reads; fake-fetch and real-API tests; a minimal one-ConfigMap page accepted in Chromium on both entry points | [Request](../field-reports/third-our-identity.md#slice-2-native-editing), [example](../../examples/native-editor/README.md), [saving](../saving.md#native-editing-through-a-host-proxy) | +| Native watch resumption | A per-handle checkpoint advanced only after a consumed event or bookmark; ordinary reconnects resume the WATCH without LIST, reset or synced; 410, malformed input and unclassifiable errors discard it for a fresh snapshot; fake-fetch, Chromium (both entry points) and real-API resume/expiry tests | [Request](../field-reports/third-our-identity.md#native-watch-resumption), [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 @@ -42,7 +44,7 @@ before starting a later one. Deliver separately reviewable changes and use the p |---|---|---| | 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 — **completed** | See [completed work](#completed-work); acceptance evidence, browser acceptance included, is recorded in the [request](../field-reports/third-our-identity.md#acceptance-for-slice-2). No editor method changed, so proposal 0009's cleanup is unaffected. | -| 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. | +| 3 | Native reconnect efficiency — **resumption completed** | See [completed work](#completed-work). Streaming-list initialization and pagination are evaluated [below](#3-native-streaming-lists-and-pagination) and remain open; implement them only with their own membership, checkpoint and recovery tests. | | 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. | | 5 | Save progress during suppressed churn | Independent evaluation using existing real-API fixtures; bounded submitted-intent recovery first, optional version delivery second. Implementation depends on measured benefit and reviewed scope. | | 6 | Comparative frontend and measurements | Follow working native viewing and editing. Compare native/full/spec and shared/unshared workloads; the larger project is outside slice 1. | @@ -53,11 +55,11 @@ before starting a later one. Deliver separately reviewable changes and use the p The native viewer and native editing are in place: native frame parsing stays separate from SSE, both connectors share the lifecycle and HTTP helpers, and native writes go back through the host -proxy as conditional merge patches. The connector re-lists on every reconnect and refuses paginated -responses; it claims no resume efficiency. Choose the next change from order 3 or from the -independent tracks; none of them waits for another. +proxy as conditional merge patches. Ordinary reconnects resume the WATCH from a per-handle +checkpoint; expiry and malformed input fall back to a fresh snapshot. The connector still requires an +unpaginated LIST. Choose the next change from the independent tracks; none of them waits for another. -Native reconnect efficiency (order 3) needs its own membership and checkpoint tests. Keep the current +Streaming lists and pagination (order 3) need their 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. Gateway continuation and save measurements proceed independently. Retain the existing @@ -87,6 +89,30 @@ fresh deliberate save after churn stops; they do not establish usable save progr churn. `task test-real-api` runs these cases. Record actual runs, server versions, skips and final commit separately from fake-client and general CI results. +## 3. Native streaming lists and pagination + +**Status: evaluated, not implemented.** Resumption removes the LIST from ordinary reconnects; these +two would reduce the cost of the snapshots that remain (first connection, expiry, malformed input). + +**Streaming-list initialization** (`sendInitialEvents=true`, `resourceVersionMatch=NotOlderThan`) +replaces LIST + WATCH with one request and avoids buffering the collection into one JSON body, which +helps memory and time-to-first-row for large collections. `reset` precedes the first synthetic ADDED; +the only boundary is the BOOKMARK annotated `k8s.io/initial-events-end: "true"` (F1), and `synced`, +pruning and arming the checkpoint (to that bookmark's resourceVersion) happen there and nowhere else. +Anything before it, a 410 included, leaves no checkpoint and prunes nothing, and synthetic ADDEDs must +not advance a resumable checkpoint. Aggregated APIs refuse the request (F6), so a handle needs a +classified, non-terminal fallback to LIST-then-WATCH. Tests: boundary gating, a partial initial +stream, a missing end bookmark, fallback classification, and the real API on both kube-apiserver and +the aggregated API. + +**Pagination** (`limit`/`continue`) bounds each response for very large collections. `reset` precedes +the first page, every page's items count as membership, nothing is pruned before the last page, and +`synced` and the checkpoint arm only after the final page and an accepted WATCH from the first page's +resourceVersion. An expired continue token (410 mid-list) abandons the whole cycle without pruning +and restarts from page one; duplicate UIDs across pages are malformed; no WATCH opens before the last +page. Tests: abort between pages, 410 on page N, a terminal error on a later page, and the store +keeping its previous membership throughout. + ## 4. Measured upstream continuation Implement behind [the Kubernetes backend](../../gateway/kube/backend.go) and its existing `Watcher` diff --git a/docs/saving.md b/docs/saving.md index 1e9f90d..b171ba4 100644 --- a/docs/saving.md +++ b/docs/saving.md @@ -140,7 +140,7 @@ version precondition, not necessarily a disagreement at an editable field. | Secret token rotates | Under `krm-full/v1` or `krm-spec/v1`, an update with a higher redaction revision | The token remains withheld; its change is visible | The RV labels are illustrative opaque strings. Neither RV nor event `seq` provides resume: a new -connection starts a complete snapshot. The +gateway connection starts a complete snapshot. The [normative contract](../spec/v1.md#6-ordering-delivery--the-state-guarantee) defines the exact guarantee. @@ -217,6 +217,17 @@ the watch delivers the same object as an ordinary event. Native objects have no to the object, status included, reaches the store with its version. A 409 still means another write landed between capture and PATCH, or that the object was replaced under the same name. +A native connection that drops resumes its WATCH from the last event the store applied, without a +new snapshot ([native connections](../packages/krm-stream/README.md#native-connections)): drafts and +conflicts stay, and changes made meanwhile arrive as ordinary events, merged against the draft as +any live event is. Until the resumed WATCH is accepted the connection is not `live`, so the example +editor answers `recovering` and writes nothing; a write captured before the drop is still guarded by +its `resourceVersion` precondition. An expired history (410) re-lists instead, and a draft survives +that snapshot while its UID does. A resume starts no snapshot, so unlike a snapshot it does not +invalidate a guarded read captured before it; an event it delivers for that object does, exactly as +on a live watch. As on a live watch, an accepted read can be followed by replayed events older than +the object it returned, until the watch catches up. + ## Creating and deleting whole objects A create and a delete are host writes exactly as a save is: the client stages the *intent*, your diff --git a/docs/why-a-gateway.md b/docs/why-a-gateway.md index 1bd3e92..214d283 100644 --- a/docs/why-a-gateway.md +++ b/docs/why-a-gateway.md @@ -62,11 +62,12 @@ The fetch connector exposes `connecting`, `syncing`, `live`, `retrying`, `closed `exhausted`. Apply state events synchronously before rendering completion. Retryable failures recover 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. +classifies HTTP and in-stream 410 (expired watch history) as recoverable by a fresh LIST instead, +and resumes its WATCH from its own checkpoint after an ordinary interruption. Dispose subscriptions and close the handle when their owner leaves. See the [client lifecycle reference](../packages/krm-stream/README.md#connections). -A new browser connection receives a fresh snapshot. SSE sequence numbers and object versions are not +A new gateway connection receives a fresh snapshot. SSE sequence numbers and object versions are not browser resume tokens. Proposed gateway upstream continuation can reduce resnapshots without changing that browser contract; see [proposal 0006](proposals/0006-stream-and-save-implementation-plan.md#4-measured-upstream-continuation). @@ -96,11 +97,15 @@ both can carry errors. Both connectors handle refusal, expiry, cancellation and 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 reads with 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`, 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](saving.md#native-editing-through-a-host-proxy) writes conditional merge patches back -through the same proxy. Resume, streaming lists and pagination follow separately. +resourceVersion. It uses the shared lifecycle and `LiveResourceStore`, and is `live` only once the +snapshot is applied and the WATCH is accepted. After that complete snapshot, an EOF, a network +failure or a transient refusal resumes the WATCH from the handle's checkpoint — the last event the +store applied, or the last bookmark — without a LIST or a new snapshot, so the store keeps its +members, drafts and conflicts and receives the missed changes as events. HTTP/in-stream 410 +discards the checkpoint and recovers with a fresh LIST within the bounded policy; terminal auth +refusal never selects another source. A paginated LIST response is refused rather than marked +complete. [Native editing](saving.md#native-editing-through-a-host-proxy) writes conditional merge +patches back through the same proxy. 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 diff --git a/examples/native-editor/README.md b/examples/native-editor/README.md index 2285459..48f3af2 100644 --- a/examples/native-editor/README.md +++ b/examples/native-editor/README.md @@ -71,6 +71,10 @@ Add `entry=bundle` to load the single-file build. - Save is enabled only while the watch is `live`, no request is in flight and no conflict is open. Typing continues during a save, and inputs keep focus and caret through watch events and reads. +- A dropped or ended watch resumes from the last event applied, without a new snapshot: the draft + and any conflict stay, Save is disabled until the resumed watch is accepted, and changes made + meanwhile are merged in as ordinary watch events. After a 410 (expired history) the page lists + again; Save stays disabled until that snapshot's watch is accepted, and the draft survives it. - `saved` waits for the watch. The next version it delivers includes the write; fields still dirty then were typed after Save or not kept by the server. With no echo within `echoWaitMs`, the page offers **Confirm current state**, which calls `editor.confirm()`. A network failure or a 5xx @@ -103,7 +107,8 @@ the page and plays the host proxy and API server, with a watch that stays open w save and its echo, typing while a save is in flight, 409 recovery and a later deliberate Save, explicit conflict resolution, a lost response and a 502 each settled by a GET before any further PATCH, an accepted write without an echo confirmed explicitly, refusals, multiline values, a deletion -and a UID replacement each with copy-out, and disconnect. +and a UID replacement each with copy-out, a draft kept through a resumed watch and through the +re-list after an expiry, and disconnect. ## What it protects diff --git a/examples/native-viewer/README.md b/examples/native-viewer/README.md index 2d6a666..06611e9 100644 --- a/examples/native-viewer/README.md +++ b/examples/native-viewer/README.md @@ -30,14 +30,18 @@ browser never holds a Kubernetes credential. - 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. +- After the first snapshot, a dropped or ended watch resumes from the last event applied — or the + last bookmark — without listing again: the History line shows `retrying → connecting → live`, + never `syncing`, and what changed while disconnected arrives as ordinary events, deletions and + selector exits included. When the API server reports the history expired (410), the page lists + again, and a resource deleted or relabelled out of the selector meanwhile disappears when that + 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. To edit through the same proxy, add the - [native editor](../native-editor/README.md), which has its own page for one ConfigMap. 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). + [native editor](../native-editor/README.md), which has its own page for one ConfigMap. Streaming + lists 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/vanilla-browser/tests/native-editor.spec.ts b/examples/vanilla-browser/tests/native-editor.spec.ts index 7057032..fdeb460 100644 --- a/examples/vanilla-browser/tests/native-editor.spec.ts +++ b/examples/vanilla-browser/tests/native-editor.spec.ts @@ -55,6 +55,10 @@ class Host { #rv = 1; #watches = new Set(); #held: string[] = []; + // Every frame by the version it carries, so a WATCH resumed from a version replays what followed. + #log: { rv: number; frame: string }[] = []; + #expireNext = false; + #holdNext: ReturnType | undefined; #server: Server; url = ""; @@ -115,6 +119,33 @@ class Host { for (const frame of this.#held.splice(0)) for (const watch of this.#watches) watch.write(frame); } + /** End every open watch, as the API server routinely does. */ + endWatches() { + // Forgotten at once, so nothing is written after the end before its close event. + for (const watch of this.#watches) { + this.#watches.delete(watch); + watch.end(); + } + } + + /** The next WATCH is accepted and then reports that its history expired, as F3 observed. */ + expireNextWatch() { + this.#expireNext = true; + } + + /** The next WATCH that is not expired waits for the returned gate before it is accepted. */ + holdNextWatch() { + this.#holdNext = gate(); + return this.#holdNext; + } + + /** The requests the page made for the collection: `list`, or `watch@`. */ + collectionCalls() { + return this.calls + .filter((c) => c.path === COLLECTION) + .map((c) => (c.query.get("watch") === "1" ? `watch@${c.query.get("resourceVersion")}` : "list")); + } + /** The API server's PATCH: preconditions checked, then the merge patch applied and echoed. */ patch = (call: Call, res: ServerResponse, { deliver = true } = {}) => { const body = call.body as { metadata?: Record }; @@ -141,6 +172,7 @@ class Host { #emit(type: string, object: ConfigMap, deliver: boolean) { const frame = `${JSON.stringify({ type, object: typed(object) })}\n`; + this.#log.push({ rv: Number(object.metadata.resourceVersion), frame }); if (!deliver) return void this.#held.push(frame); for (const watch of this.#watches) watch.write(frame); } @@ -163,9 +195,22 @@ class Host { if (call.path === COLLECTION && call.method === "GET") { expect(call.query.get("fieldSelector")).toBe("metadata.name=settings"); if (call.query.get("watch") === "1") { - // Accepted, and open until the page or the test closes it. + if (this.#expireNext) { + this.#expireNext = false; + res.writeHead(200, { "Content-Type": "application/json", Connection: "close" }); + const status = { kind: "Status", apiVersion: "v1", status: "Failure", code: 410, reason: "Expired" }; + const message = `too old resource version: ${call.query.get("resourceVersion")} (${this.#rv})`; + return void res.end(`${JSON.stringify({ type: "ERROR", object: { ...status, message } })}\n`); + } + const held = this.#holdNext; + this.#holdNext = undefined; + await held?.opened; + // Accepted, and open until the page or the test closes it. A watch from an older version + // first replays what followed it, as the API server does from its history. res.writeHead(200, { "Content-Type": "application/json" }); res.flushHeaders(); + const from = Number(call.query.get("resourceVersion")); + for (const { rv, frame } of this.#log) if (rv > from && !this.#held.includes(frame)) res.write(frame); this.#watches.add(res); res.on("close", () => this.#watches.delete(res)); return; @@ -614,3 +659,46 @@ test("a deleted object shows its unsaved edits at once, with nothing to open in await expect(save(page)).toBeDisabled(); expect(host.objectCalls()).toEqual([]); }); + +test("a draft survives a resumed watch, and saving waits while an expired watch re-lists", async ({ + page, + entry, + host, +}) => { + await open(page, host, entry); + await field(page, "data value").fill("mine"); + + // The watch ends, and another writer changes a field the person is not editing before the page + // reconnects. The page resumes from its checkpoint: the change is replayed, with no new LIST. + host.endWatches(); + host.change((o) => { + o.metadata.annotations.owner = "team-c"; + }); + await expect(field(page, "annotation owner")).toHaveValue("team-c"); + await expect(page.locator("#state")).toHaveText("live"); + expect(host.collectionCalls()).toEqual(["list", "watch@1", "watch@1"]); + await expect(field(page, "data value")).toHaveValue("mine"); + await expect(page.locator("#changes li")).toHaveCount(1); + await expect(page.locator("#changes")).toContainText('change data value: "base" → "mine"'); + + // The next resume finds the history expired. Until the replacement snapshot's watch is accepted, + // the edit stays in the form and Save stays disabled. + host.expireNextWatch(); + const accepted = host.holdNextWatch(); + host.endWatches(); + await expect(page.locator("#error")).toHaveText("RESYNC_REQUIRED: too old resource version: 2 (2)"); + await expect(page.locator("#state")).toHaveText("syncing"); + await expect(save(page)).toBeDisabled(); + await expect(page.locator("#hint")).toHaveText("Saving resumes once the watch is live; your edits stay here."); + await expect(field(page, "data value")).toHaveValue("mine"); + expect(host.collectionCalls()).toEqual(["list", "watch@1", "watch@1", "watch@2", "list", "watch@2"]); + expect(host.objectCalls(), "nothing is written while recovering").toEqual([]); + + accepted.open(); + await expect(page.locator("#state")).toHaveText("live"); + await expect(save(page)).toBeEnabled(); + await save(page).click(); + await expect(outcome(page)).toHaveAttribute("data-kind", "echoed"); + expect(host.patchBodies()).toEqual([{ metadata: { uid: "u1", resourceVersion: "2" }, data: { value: "mine" } }]); + expect(host.object!.metadata.annotations.owner).toBe("team-c"); +}); diff --git a/examples/vanilla-browser/tests/native-viewer.spec.ts b/examples/vanilla-browser/tests/native-viewer.spec.ts index 89f3f55..7023a87 100644 --- a/examples/vanilla-browser/tests/native-viewer.spec.ts +++ b/examples/vanilla-browser/tests/native-viewer.spec.ts @@ -42,7 +42,7 @@ function gate() { return { opened, open }; } -test("the native viewer lists, goes live once the watch is accepted, re-lists and disconnects", async ({ +test("the native viewer lists, goes live on acceptance, resumes after EOF, re-lists after expiry and disconnects", async ({ page, entry, }) => { @@ -53,8 +53,38 @@ test("the native viewer lists, goes live once the watch is accepted, re-lists an 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 bDeleted = item("b", "uid-b", "21", "one"); const c = item("c", "uid-c", "12", "new"); - const watches = [gate(), gate()]; + const bookmark = { + type: "BOOKMARK", + object: { kind: "ConfigMap", apiVersion: "v1", metadata: { resourceVersion: "20" } }, + }; + const expired = { + type: "ERROR", + object: { + kind: "Status", + apiVersion: "v1", + status: "Failure", + code: 410, + reason: "Expired", + message: "too old resource version: 20 (25)", + }, + }; + // Each WATCH the page sends: the resourceVersion it must ask for, and what it is answered once + // its gate opens. A fulfilled body ends, so every answered watch is followed by EOF. + const script = [ + // The snapshot's watch: two changes and a bookmark, then EOF. The page resumes from the bookmark. + { + from: "10", + gate: gate(), + body: frames({ type: "MODIFIED", object: typed(a2) }, { type: "ADDED", object: typed(c) }, bookmark), + }, + // The resumed watch replays b's deletion, then the history expires. + { from: "20", gate: gate(), body: frames({ type: "DELETED", object: typed(bDeleted) }, expired) }, + // The replacement snapshot's watch, then a resume from its list version that is never answered. + { from: "30", gate: gate(), body: "" }, + { from: "30", gate: undefined, body: "" }, + ]; let lists = 0; let watchCount = 0; await page.route("**/k8s/**", async (route) => { @@ -63,31 +93,25 @@ test("the native viewer lists, goes live once the watch is accepted, re-lists an 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. + // The first snapshot holds a and b. The second holds only a: c was deleted while the history + // was expired, and nothing reports it except that complete snapshot. const items = lists === 1 ? [a1, b] : [a2]; await route.fulfill({ - json: { kind: "ConfigMapList", apiVersion: "v1", metadata: { resourceVersion: `${lists}0` }, items }, + json: { + kind: "ConfigMapList", + apiVersion: "v1", + metadata: { resourceVersion: `${lists === 1 ? 1 : 3}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) }, - ) - : "", - }); + const step = script[watchCount++]; + expect(url.searchParams.get("resourceVersion")).toBe(step?.from); + expect(url.searchParams.get("allowWatchBookmarks")).toBe("true"); + if (!step?.gate) return; // never answered: the page waits, connecting, until it disconnects + await step.gate.opened; + await route.fulfill({ contentType: "application/json", body: step.body }); }); await page.goto(`/examples/native-viewer/index.html?entry=${entry}&namespace=app&labelSelector=tier%3Dweb`); @@ -101,22 +125,42 @@ test("the native viewer lists, goes live once the watch is accepted, re-lists an await expect(state).toHaveText("syncing"); await expect(history).not.toContainText("live"); - watches[0]!.open(); + script[0]!.gate!.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-c")).toBeVisible(); + // The watch ended. The page resumes it from the bookmark instead of listing again: while the + // resumed watch is opening it is connecting, never syncing, and it keeps every row. + await expect.poll(() => watchCount).toBe(2); + await expect(state).toHaveText("connecting"); + expect(lists, "a resume is not a snapshot").toBe(1); + await expect(row("uid-b")).toBeVisible(); + + script[1]!.gate!.open(); + // Accepted: live at once, and the missed deletion arrives. Then the history expires. await expect(row("uid-b")).toHaveCount(0); + await expect(page.locator("#error")).toHaveText("RESYNC_REQUIRED: too old resource version: 20 (25)"); + await expect.poll(() => lists).toBe(2); + // The replacement snapshot is incomplete until its watch is accepted, so c — missing from it — is + // still shown, and the page is syncing. + await expect(state).toHaveText("syncing"); await expect(row("uid-c")).toBeVisible(); - watches[1]!.open(); - // The replacement snapshot completes: c is pruned. + script[2]!.gate!.open(); + // The replacement snapshot completes: c is pruned. Its watch ends, and the next one resumes. 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 expect.poll(() => watchCount).toBe(4); + await expect(state).toHaveText("connecting"); + await expect(history).toHaveText( + [ + ...["connecting", "syncing", "syncing", "live"], + ...["retrying", "connecting", "live"], + ...["retrying", "connecting", "syncing", "syncing", "live"], + ...["retrying", "connecting"], + ].join(" → "), + ); + expect(lists).toBe(2); await page.getByRole("button", { name: "Disconnect" }).click(); await expect(state).toHaveText("closed"); diff --git a/gateway/kube/native_e2e_test.go b/gateway/kube/native_e2e_test.go index d4e7134..1cc364b 100644 --- a/gateway/kube/native_e2e_test.go +++ b/gateway/kube/native_e2e_test.go @@ -32,6 +32,7 @@ import ( "github.com/ConfigButler/krm-stream/gateway" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" "k8s.io/client-go/rest" ) @@ -45,6 +46,124 @@ type nativeHost struct { // competing write between the browser capturing its save and Kubernetes checking it. beforePatch func() patches []int // the API server's answer to each forwarded PATCH + + // Test-only watch controls for the resume case. They change nothing until a test uses them. + gets []forwardedGet // every forwarded GET, in order + watches map[int]context.CancelFunc // the open WATCH connections, by index into gets + hold chan struct{} // non-nil: the next WATCH waits until it is closed + held chan struct{} // closed once the next WATCH is being held + expire int // how many of the next WATCHes are rewritten to resourceVersion=1 +} + +// forwardedGet is one GET the host forwarded: a LIST, a WATCH or the editor's object read. +type forwardedGet struct { + Path string + Watch bool + // Asked is the resourceVersion the browser asked a WATCH to start from; Sent is what the host + // forwarded, which differs only when the test induced an expiry. + Asked, Sent string + // Answer is the API server's HTTP status. A WATCH whose history expired is still answered 200: + // the 410 arrives as an ERROR event in the stream. + Answer int +} + +type forwardedKey struct{} + +// lists counts the forwarded LISTs and WATCHes of one collection path. +func (h *nativeHost) lists(path string) (lists, watches int) { + h.mu.Lock() + defer h.mu.Unlock() + for _, g := range h.gets { + switch { + case g.Path != path: + case g.Watch: + watches++ + default: + lists++ + } + } + return lists, watches +} + +// forwardedWatches returns every forwarded WATCH, in order. +func (h *nativeHost) forwardedWatches() []forwardedGet { + h.mu.Lock() + defer h.mu.Unlock() + return slices.DeleteFunc(slices.Clone(h.gets), func(g forwardedGet) bool { return !g.Watch }) +} + +// dropWatches cuts every open WATCH connection, as a network failure would, and says how many. +func (h *nativeHost) dropWatches() int { + h.mu.Lock() + defer h.mu.Unlock() + for _, cancel := range h.watches { + cancel() + } + return len(h.watches) +} + +// holdNextWatch makes the next WATCH wait, so the test can change the cluster while the browser is +// disconnected. held is closed once that WATCH arrives; release lets it through. +func (h *nativeHost) holdNextWatch() (held <-chan struct{}, release func()) { + h.mu.Lock() + defer h.mu.Unlock() + hold := make(chan struct{}) + h.hold, h.held = hold, make(chan struct{}) + return h.held, sync.OnceFunc(func() { close(hold) }) +} + +// expireNextWatches rewrites the resourceVersion of the next n WATCHes to "1", long compacted away, +// so the REAL API server answers each with its own in-stream 410 Expired. The expiry is induced; the +// 410 is the server's. +func (h *nativeHost) expireNextWatches(n int) { + h.mu.Lock() + defer h.mu.Unlock() + h.expire = n +} + +// forwardGet applies the watch controls to one GET and records it. It returns the request to forward +// and a function to call once the response is over. +func (h *nativeHost) forwardGet(r *http.Request) (*http.Request, func()) { + query := r.URL.Query() + watch := query.Get("watch") == "1" || query.Get("watch") == "true" + if watch { + h.mu.Lock() + hold, held := h.hold, h.held + h.hold, h.held = nil, nil + h.mu.Unlock() + if hold != nil { + close(held) + select { + case <-hold: + case <-r.Context().Done(): + } + } + } + h.mu.Lock() + defer h.mu.Unlock() + get := forwardedGet{Path: r.URL.Path, Watch: watch} + if watch { + get.Asked = query.Get("resourceVersion") + get.Sent = get.Asked + if h.expire > 0 { + h.expire-- + get.Sent = "1" + query.Set("resourceVersion", get.Sent) + r.URL.RawQuery = query.Encode() + } + } + index := len(h.gets) + h.gets = append(h.gets, get) + ctx, cancel := context.WithCancel(context.WithValue(r.Context(), forwardedKey{}, index)) + if watch { + h.watches[index] = cancel + } + return r.WithContext(ctx), func() { + cancel() + h.mu.Lock() + defer h.mu.Unlock() + delete(h.watches, index) + } } func (h *nativeHost) nextPatch(hook func()) { @@ -70,7 +189,7 @@ func nativeProxy(t *testing.T, writable bool) *nativeHost { if err != nil { t.Fatalf("API server address: %v", err) } - host := &nativeHost{} + host := &nativeHost{watches: map[int]context.CancelFunc{}} proxy := &httputil.ReverseProxy{ Rewrite: func(r *httputil.ProxyRequest) { r.Out.URL.Scheme = target.Scheme @@ -83,10 +202,13 @@ func nativeProxy(t *testing.T, writable bool) *nativeHost { // A watch is a long response; every frame must reach the browser as it is written. FlushInterval: -1, ModifyResponse: func(res *http.Response) error { + host.mu.Lock() + defer host.mu.Unlock() if res.Request.Method == http.MethodPatch { - host.mu.Lock() host.patches = append(host.patches, res.StatusCode) - host.mu.Unlock() + } + if index, ok := res.Request.Context().Value(forwardedKey{}).(int); ok { + host.gets[index].Answer = res.StatusCode } return nil }, @@ -98,6 +220,11 @@ func nativeProxy(t *testing.T, writable bool) *nativeHost { } switch { case r.Method == http.MethodGet: + // A dropped WATCH ends with its upstream request cancelled mid-body; the proxy then aborts + // the browser's connection, exactly as a network failure would. + forward, done := host.forwardGet(r) + defer done() + r = forward case r.Method == http.MethodPatch && writable: // Only a JSON merge patch: a JSON Patch or an apply patch has semantics the check below // does not cover. @@ -176,6 +303,12 @@ type nativeDriver struct { stdin io.WriteCloser lines chan nativeLine members []nativeMember + // What await has read so far: each state in order, how often each state event arrived, and the + // connector errors a test declared it expects (by code, with how many more are allowed). + states []string + events map[string]int + expected map[string]int + errors []string } // newNativeDriver starts the browser side on the collection that scope names behind proxy. @@ -202,7 +335,9 @@ func newNativeDriver(t *testing.T, proxy string, scope map[string]string) *nativ if err := cmd.Start(); err != nil { t.Fatal(err) } - d := &nativeDriver{t: t, stdin: stdin, lines: make(chan nativeLine, 256)} + d := &nativeDriver{ + t: t, stdin: stdin, lines: make(chan nativeLine, 256), events: map[string]int{}, expected: map[string]int{}, + } go func() { defer close(d.lines) scanner := bufio.NewScanner(stdout) @@ -221,7 +356,12 @@ func newNativeDriver(t *testing.T, proxy string, scope map[string]string) *nativ return d } -// await reads lines until pred holds, failing on any error the connector reports. +// expect allows n more connector errors with this code; any other error still fails the test. +func (d *nativeDriver) expect(code string, n int) { + d.expected[code] += n +} + +// await reads lines until pred holds, failing on any connector error the test did not expect. func (d *nativeDriver) await(what string, pred func(nativeLine) bool) nativeLine { d.t.Helper() deadline := time.After(30 * time.Second) @@ -232,13 +372,21 @@ func (d *nativeDriver) await(what string, pred func(nativeLine) bool) nativeLine 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.Error.Terminal || d.expected[line.Error.Code] == 0 { + d.t.Fatalf("%s: connector error %s: %s (terminal %v)", what, line.Error.Code, line.Error.Message, line.Error.Terminal) + } + d.expected[line.Error.Code]-- + d.errors = append(d.errors, line.Error.Code+": "+line.Error.Message) } if line.CommandError != "" { d.t.Fatalf("%s: command failed: %s", what, line.CommandError) } + if line.State != "" { + d.states = append(d.states, line.State) + } if line.Event != "" { d.members = line.Members + d.events[line.Event]++ } if pred(line) { return line @@ -267,51 +415,74 @@ func (d *nativeDriver) member(name string) nativeMember { return nativeMember{} } -func TestRealAPINativeWatchThroughHostProxy(t *testing.T) { +// configMapFixture changes labelled ConfigMaps in a scratch namespace, failing the test on error. +type configMapFixture struct { + t *testing.T + ctx context.Context + api typedcorev1.ConfigMapInterface +} + +func newConfigMapFixture(t *testing.T) (*configMapFixture, string) { + t.Helper() 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 + return &configMapFixture{t: t, ctx: context.Background(), api: cs.CoreV1().ConfigMaps(ns)}, ns +} + +func (f *configMapFixture) create(name, tier, value string) *corev1.ConfigMap { + f.t.Helper() + created, err := f.api.Create(f.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 { + f.t.Fatalf("create %s: %v", name, err) } - 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) - } + return created +} + +func (f *configMapFixture) get(name string) *corev1.ConfigMap { + f.t.Helper() + current, err := f.api.Get(f.ctx, name, metav1.GetOptions{}) + if err != nil { + f.t.Fatalf("get %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) - } + return current +} + +func (f *configMapFixture) update(name string, change func(*corev1.ConfigMap)) { + f.t.Helper() + current := f.get(name) + change(current) + if _, err := f.api.Update(f.ctx, current, metav1.UpdateOptions{}); err != nil { + f.t.Fatalf("update %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) +} + +func (f *configMapFixture) remove(name string) { + f.t.Helper() + if err := f.api.Delete(f.ctx, name, metav1.DeleteOptions{}); err != nil { + f.t.Fatalf("delete %s: %v", name, err) + } +} + +// hasNames holds for a state event after which the store's members are exactly want, by name. +func hasNames(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) } +} + +func TestRealAPINativeWatchThroughHostProxy(t *testing.T) { + fx, ns := newConfigMapFixture(t) + create, update, remove := fx.create, fx.update, fx.remove create("a", "web", "1") create("b", "web", "1") @@ -544,3 +715,156 @@ func TestRealAPINativeEditThroughHostProxy(t *testing.T) { } d.await("closed", func(line nativeLine) bool { return line.State == "closed" }) } + +// waitClosed waits for ch to be closed. +func waitClosed(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(30 * time.Second): + t.Fatalf("%s: not reached", what) + } +} + +// The browser resumes a dropped WATCH from its checkpoint through the same host proxy instead of +// listing again: what changed while it was disconnected — an update, a delete, a selector exit and a +// same-name replacement — arrives as watch events, with no new LIST and no reset, and a draft made +// before the drop survives and saves. Then the proxy rewrites the next WATCH's resourceVersion to +// "1": the expiry is induced, but the 410 is the real API server's own in-stream answer, and the +// browser recovers with a fresh LIST — once, then with two expiries in a row. +func TestRealAPINativeResumeThroughHostProxy(t *testing.T) { + fx, ns := newConfigMapFixture(t) + fx.create("a", "web", "1") + fx.create("b", "web", "1") + fx.create("c", "db", "1") + fx.create("d", "web", "1") + original := fx.create("r", "web", "first") + + proxy := nativeProxy(t, true) + collection := "/k8s/api/v1/namespaces/" + ns + "/configmaps" + d := newNativeDriver(t, proxy.URL+"/k8s", map[string]string{ + "version": "v1", "resource": "configmaps", "namespace": ns, "labelSelector": "tier=web", + }) + counts := func(when string, wantLists, wantWatches int) { + t.Helper() + if lists, watches := proxy.lists(collection); lists != wantLists || watches != wantWatches { + t.Fatalf("%s: want %d LISTs and %d WATCHes forwarded, got %d and %d", when, wantLists, wantWatches, lists, watches) + } + } + started := time.Now() + d.await("live", func(line nativeLine) bool { return line.State == "live" }) + if got := d.names(); !slices.Equal(got, []string{"a", "b", "d", "r"}) { + t.Fatalf("snapshot: want a, b, d and r, got %v", got) + } + counts("snapshot", 1, 1) + + // A draft made before the interruption, on a field nobody else writes. + d.set("a", []any{"data", "note"}, "mine") + + // Drop the watch, and hold its resumption while the cluster changes underneath. + held, release := proxy.holdNextWatch() + if n := proxy.dropWatches(); n != 1 { + t.Fatalf("dropped %d watches, want 1", n) + } + waitClosed(t, held, "the resumed watch") + mark := len(d.states) + resumeStarted := time.Now() + fx.update("a", func(cm *corev1.ConfigMap) { cm.Data["value"] = "2" }) + fx.remove("b") + // A label change that takes d out of the selector arrives as a deletion. + fx.update("d", func(cm *corev1.ConfigMap) { cm.Labels["tier"] = "db" }) + fx.remove("r") + replacement := fx.create("r", "web", "second") + release() + d.await("missed changes replayed", func(line nativeLine) bool { + return hasNames("a", "r")(line) && line.Members[0].Data["value"] == "2" && + line.Members[1].UID == string(replacement.UID) + }) + resumeTook := time.Since(resumeStarted) + if original.UID == replacement.UID { + t.Fatal("the recreated object kept its UID; the case proves nothing") + } + counts("resume", 1, 2) + if resumed := proxy.forwardedWatches()[1]; resumed.Asked == "" || resumed.Sent != resumed.Asked || + resumed.Answer != http.StatusOK { + t.Fatalf("the resumed watch: %+v", resumed) + } + if d.events["reset"] != 1 || d.events["synced"] != 1 { + t.Fatalf("a resumed watch starts no snapshot: events %v", d.events) + } + if got := d.states[mark:]; !slices.Equal(got, []string{"retrying", "connecting", "live"}) { + t.Fatalf("states while resuming: want retrying, connecting, live; got %v", got) + } + + // The draft survived the resume, and saves at the version the resumed watch delivered. + if got := d.save("a"); got != "saved" { + t.Fatalf("save after the resume: want saved, got %s", got) + } + d.holds("save echoed", func(members []nativeMember) bool { + return len(members) > 0 && members[0].Name == "a" && members[0].Data["note"] == "mine" + }) + if cm := fx.get("a"); cm.Data["value"] != "2" || cm.Data["note"] != "mine" { + t.Fatalf("after the save: want value 2 and note mine, got %v", cm.Data) + } + + // Expiry: the resumed WATCH asks for a resourceVersion long compacted away. The API server + // accepts it and reports 410 in the stream; the browser discards its checkpoint and lists. + proxy.expireNextWatches(1) + held, release = proxy.holdNextWatch() + mark = len(d.states) + expiryStarted := time.Now() + proxy.dropWatches() + waitClosed(t, held, "the expiring watch") + fx.create("f", "web", "1") + release() + d.expect("RESYNC_REQUIRED", 1) + d.await("expiry recovered", func(line nativeLine) bool { return d.events["reset"] == 2 && line.State == "live" }) + expiryTook := time.Since(expiryStarted) + if got := d.names(); !slices.Equal(got, []string{"a", "f", "r"}) { + t.Fatalf("after the replacement snapshot: want a, f and r, got %v", got) + } + counts("expiry", 2, 4) + expiredWatch := proxy.forwardedWatches()[2] + if expiredWatch.Sent != "1" || expiredWatch.Asked == "1" || expiredWatch.Answer != http.StatusOK { + t.Fatalf("the expired watch: want an accepted watch from resourceVersion 1, got %+v", expiredWatch) + } + if len(d.errors) != 1 { + t.Fatalf("want one RESYNC_REQUIRED, got %v", d.errors) + } + // The expired watch is accepted before its ERROR arrives, so it is briefly live; after the 410 the + // state is never live again until the replacement snapshot completes. + if got, want := d.states[mark:], []string{"retrying", "connecting", "live", "retrying", "connecting", "syncing", "syncing", "live"}; !slices.Equal(got, want) { + t.Fatalf("states through expiry: want %v, got %v", want, got) + } + + // Two expiries in a row: the resumed WATCH and the replacement snapshot's WATCH. Each one costs + // a retry with backoff, and the third snapshot converges. + proxy.expireNextWatches(2) + proxy.dropWatches() + fx.update("a", func(cm *corev1.ConfigMap) { cm.Data["value"] = "3" }) + d.expect("RESYNC_REQUIRED", 2) + d.await("repeated expiry recovered", func(line nativeLine) bool { + return d.events["reset"] == 4 && line.State == "live" + }) + d.holds("a converged", func(members []nativeMember) bool { + return len(members) == 3 && members[0].Name == "a" && members[0].Data["value"] == "3" + }) + counts("repeated expiry", 4, 7) + watches := proxy.forwardedWatches() + if watches[4].Sent != "1" || watches[5].Sent != "1" || watches[6].Sent != watches[6].Asked { + t.Fatalf("repeated expiry: want watches 5 and 6 expired and 7 not, got %+v", watches[4:]) + } + if len(d.errors) != 3 { + t.Fatalf("want three RESYNC_REQUIRED, got %v", d.errors) + } + + t.Logf("resume: %d LISTs, %d WATCHes; resumed from resourceVersion %q, replay converged in %v; "+ + "expiry recovery %v; errors %v; total %v", + 4, len(watches), proxy.forwardedWatches()[1].Asked, resumeTook.Round(time.Millisecond), + expiryTook.Round(time.Millisecond), d.errors, time.Since(started).Round(time.Millisecond)) + + 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/README.md b/packages/krm-stream/README.md index cf3d41d..8a30117 100644 --- a/packages/krm-stream/README.md +++ b/packages/krm-stream/README.md @@ -136,26 +136,48 @@ guarantees as `connectResourceStream`. `nativeCollectionURL` builds core (`/api/ `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. +`resourceVersion`, `resourceVersionMatch`, `limit`, `continue`, `sendInitialEvents` or +`allowWatchBookmarks`, which the connector controls. + +The first 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 `allowWatchBookmarks=true`, 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 are never +delivered and change no object. + +From then on the handle keeps a checkpoint: the `resourceVersion` of the last event the consumer +applied, or of the last bookmark. An EOF, a network failure, a watch cut off inside a frame, or a +retryable refusal (408, 429 or 5xx, as an HTTP status or in the stream) **resumes** the WATCH from +that checkpoint with the same selectors: no LIST, no `reset` and no `synced`. The state goes +`connecting` → `live` once the resumed WATCH is accepted, without `syncing`, and the store keeps its +members, drafts and conflicts. Whatever changed meanwhile arrives as ordinary events: updates as +`modified`, deletes and selector exits as `deleted`, a same-name object recreated with a new UID as +the old UID's `deleted` and the new one's `added`. This is sound for the same reason a live watch +is: the store holds everything up to the checkpoint and the WATCH delivers everything after it, in +order; a conditional save's `resourceVersion` precondition protects a write against anything +stale. The checkpoint is private to one handle — never exposed, accepted from a caller or shared — +so it is bound to that handle's URL, selectors and credentials. Resource versions are opaque: the +checkpoint is replaced in stream order and never compared. + +HTTP or in-stream 410 (expired watch history) is reported as `RESYNC_REQUIRED` and discards the +checkpoint, as do malformed frames and an in-stream error without a code. The next connection then +lists again, and prior state is kept until that replacement snapshot completes and prunes what it no +longer lists. An initialization interrupted before `synced` leaves no checkpoint either. While it +recovers the state is `retrying`, `connecting` or `syncing`, never `live`, so an editor gated on +`live` stays disabled. Every reconnect, resumed or not, consumes the bounded retry budget with +backoff, so 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, on a resumed WATCH too. + +This release neither paginates nor uses streaming lists (`sendInitialEvents`): 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. A resumed WATCH starts no snapshot, so a guarded read a host +captured with `captureReconciliation` before it is not invalidated by the resume, exactly as on a +live watch; only a snapshot invalidates one. 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. A store fed by `connectNativeWatch` can be edited like any other. `nativeObjectURL(proxy, { ...scope, namespace, name })` addresses one member under the same proxy and collection, for a conditional JSON diff --git a/packages/krm-stream/e2e/native-driver.ts b/packages/krm-stream/e2e/native-driver.ts index ca2336a..7b5494d 100644 --- a/packages/krm-stream/e2e/native-driver.ts +++ b/packages/krm-stream/e2e/native-driver.ts @@ -5,7 +5,8 @@ // the browser: the REAL connectNativeWatch feeding the REAL store under the default edit policy, and // the REAL native editor from examples/native-editor writing back through the same proxy, 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. +// describing what the store holds, so the Go side can wait for the cluster's changes to arrive. The +// event lines include `reset` and `synced`, so the Go side can tell a resumed watch from a snapshot. // // Commands, one per line on stdin: // @@ -56,7 +57,13 @@ const handle = connectNativeWatch( applyStreamEvent(store, event); report({ event: event.type, members: members() }); }, - { retryDelayMs: 100, onError: (code, message, terminal) => report({ error: { code, message, terminal } }) }, + { + retryDelayMs: 100, + // The resume case drops and expires watches several times within one healthy period; each + // reconnect still consumes the budget, so give it room to observe them all. + maxRetries: 20, + onError: (code, message, terminal) => report({ error: { code, message, terminal } }), + }, ); let live = false; handle.subscribe((state) => { diff --git a/packages/krm-stream/src/lifecycle.ts b/packages/krm-stream/src/lifecycle.ts index b4562f7..b5711e7 100644 --- a/packages/krm-stream/src/lifecycle.ts +++ b/packages/krm-stream/src/lifecycle.ts @@ -72,7 +72,10 @@ export interface TransportHooks { 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. */ + /** The stream is live, with the connection still open: either the consumer has applied `synced`, + * or a resumed watch was accepted and the consumer's state is already complete — it holds + * everything up to the watch's checkpoint, and the watch delivers what follows. A resumed watch + * calls neither `opened` nor `reset`, so it goes `connecting` → `live` without `syncing`. */ synced(): void; /** A missing or duplicated event. The event beyond the gap was discarded; the connection ends. */ gap(expected: number, received: number): void; diff --git a/packages/krm-stream/src/native.ts b/packages/krm-stream/src/native.ts index e1b8f64..9d10dc4 100644 --- a/packages/krm-stream/src/native.ts +++ b/packages/krm-stream/src/native.ts @@ -3,11 +3,24 @@ // 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. +// A handle's first connection, and every connection after its checkpoint is discarded, starts with +// a fresh, complete LIST: the snapshot repairs whatever a previous watch missed — deletes, selector +// exits, a UID replaced under the same name — and the store prunes only once it completes. Once that +// snapshot is applied and its WATCH accepted, the handle keeps a CHECKPOINT: the resourceVersion of +// the last event its consumer applied, or of the last bookmark. An ordinary reconnect — EOF, a +// network failure, a retryable refusal — resumes the WATCH from it instead of listing again. The +// consumer already holds everything up to the checkpoint and the resumed watch delivers everything +// after it, in order: the same invariant any live watch with events in flight relies on. So a +// resumed watch starts no snapshot — no reset, no synced — and the store keeps its membership, +// drafts and conflicts. A write is protected against replayed staleness by its resourceVersion +// precondition, exactly as on a live watch. +// +// Expiry (410), malformed input and unclassifiable errors discard the checkpoint, and the next +// connection lists again. The checkpoint belongs to one connectNativeWatch call: it is never +// exported, accepted from a caller or shared between handles, so it is bound to that handle's URL, +// selectors and credentials. ResourceVersions are opaque: the checkpoint is replaced in stream +// order, never compared, parsed or ordered. Streaming lists and pagination are later work with their own +// recovery rules; this file deliberately implements neither. // // 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 @@ -24,28 +37,47 @@ import { 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"]; +const reserved = [ + "watch", + "resourceVersion", + "resourceVersionMatch", + "limit", + "continue", + "sendInitialEvents", + "allowWatchBookmarks", +]; /** 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. + * The first 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`. Every WATCH asks for + * bookmarks; they are never delivered and change no object. + * + * After a complete snapshot the handle keeps a checkpoint: the resourceVersion of the last event the + * consumer applied, or of the last bookmark. An EOF, a network failure, a truncated final frame or a + * retryable refusal (408, 429 or 5xx, as an HTTP status or in the stream) RESUMES the WATCH from that + * checkpoint: no LIST, no `reset` and no `synced`, and the state goes `connecting` → `live` once the + * resumed watch is accepted. The consumer keeps everything it holds, drafts and conflicts included, + * and receives what changed in the meantime — deletes and selector exits as `deleted`, a same-name + * replacement as the old UID's `deleted` and the new UID's `added`. * - * 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. + * HTTP or in-stream 410 (history expired) is reported as RESYNC_REQUIRED and discards the + * checkpoint, as do malformed frames and an in-stream error without a code: the next connection + * LISTs again, and a store keeps its previous state until that snapshot completes and then prunes + * what it no longer contains. An initialization interrupted before `synced` leaves no checkpoint + * either. Every reconnect consumes the bounded retry budget, honoring `Retry-After` and a Status's + * `retryAfterSeconds`. HTTP or in-stream 401, 403 and every other 4xx are terminal, on a resumed + * watch too, 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. */ + * The checkpoint is private to the handle, so it is bound to this URL and these options: it is + * never exposed, accepted or shared. 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, @@ -56,7 +88,9 @@ export function connectNativeWatch( 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); + // This handle's own, for its whole life: no other handle, URL or identity ever reads it. + const position: Position = { checkpoint: undefined, type: undefined }; + return runConnection((hooks, signal) => watchOnce(url, hooks, signal, opts, position), consume, opts); } /** The apiVersion and kind a typed collection's items have: `ConfigMapList` lists `ConfigMap`s. */ @@ -65,17 +99,39 @@ interface ItemType { kind: string; } -/** A frame or a response this connector cannot use. Reported, never skipped: the connection ends - * and the next one re-lists. */ +/** What one handle carries from one connection to the next. Only that handle's transport reads or + * writes it. */ +interface Position { + /** Where the next WATCH resumes: the resourceVersion of the last event the consumer applied, or of + * the last bookmark, after a complete snapshot. Undefined when the next connection must LIST. + * Opaque: replaced in stream order, never compared, parsed or ordered. */ + checkpoint: string | undefined; + /** The item type the last complete LIST named, for watch objects without type metadata: a resumed + * watch has no collection of its own to read it from. */ + type: ItemType | undefined; +} + +/** A frame or a response this connector cannot use. Reported, never skipped: the connection ends, + * the checkpoint is discarded, and the next connection 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. */ +/** The watch ended inside a frame. Unlike malformed input this says nothing about what was + * delivered: the partial frame was never applied, so the checkpoint still names the last event that + * was, and the next connection resumes from it. */ +class Truncated extends Error {} + +/** Whether a refusal with this HTTP status, on the response or in the stream, leaves the checkpoint + * usable: a transient condition of the server, not of the position. */ +const resumable = (status: number) => status === 408 || status === 429 || status >= 500; + +/** One connection: LIST, then WATCH until it ends — or, with a checkpoint, only the WATCH from it. + * Resolves in every case, after the readers and listeners are released. */ async function watchOnce( url: string, hooks: TransportHooks, signal: AbortSignal, opts: ResourceStreamOptions, + position: Position, ): Promise { const controller = new AbortController(); const abort = () => controller.abort(); @@ -85,60 +141,88 @@ async function watchOnce( // Called detached, so a consumer never sees this hooks object as `this`. const { consume } = hooks; - /** Report a refused response. The connection ends either way. */ + /** Report a refused response. The connection ends either way; only a transient refusal keeps the + * checkpoint, so an expired or refused position is never asked for again. */ const refused = async (phase: string, res: Response) => { + if (!resumable(res.status)) position.checkpoint = undefined; 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) => { + // Something was delivered that cannot be trusted, or something that should have been was not: + // the position is unknown, and only a fresh snapshot repairs it. + position.checkpoint = undefined; if (!aborted()) hooks.error("INTERNAL", `native ${phase}: ${error.message}`, false); }; + /** The consumer applied an event, or the stream passed a bookmark: the next connection resumes + * after it. An event without a usable resourceVersion leaves no position to resume from. Nothing + * moves once the stream is closing, which is also how a consumer exception leaves it unmoved. */ + const advance = (resourceVersion: unknown) => { + if (aborted()) return; + position.checkpoint = typeof resourceVersion === "string" && resourceVersion !== "" ? resourceVersion : undefined; + }; 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(); + // Read once: this connection resumes or re-lists, never both. + const resumeFrom = position.checkpoint; + let from: string; + if (resumeFrom === undefined) { + 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; + 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 } - if (error instanceof Malformed) return malformed("list", error); - throw error; // the network failed while the body arrived - } + position.type = collection.type; + from = collection.resourceVersion; - // 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) { + // Membership is established only by a COMPLETE collection, so nothing above this line delivers. + hooks.reset(); if (aborted()) return; - consume({ type: "added", object }); + consume({ type: "reset" }); + for (const object of collection.items) { + if (aborted()) return; + consume({ type: "added", object }); + } + if (aborted()) return; + } else { + from = resumeFrom; } - if (aborted()) return; + // Bookmarks keep a quiet collection's checkpoint recent enough to resume from. const watched = await request( - `${url}${url.includes("?") ? "&" : "?"}watch=1&resourceVersion=${encodeURIComponent(collection.resourceVersion)}`, + `${url}${url.includes("?") ? "&" : "?"}watch=1&allowWatchBookmarks=true&resourceVersion=${encodeURIComponent(from)}`, "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(() => {})); + if (resumeFrom === undefined) { + // 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`, + // and an initialization that never completed leaves no checkpoint: the next connection lists. + if (aborted()) return void (await watched.body.cancel().catch(() => {})); + position.checkpoint = from; + } + // A resumed watch starts no snapshot: the consumer already holds everything up to the checkpoint + // and this watch delivers everything after it, in order. Its acceptance alone makes it live. hooks.synced(); const reader = watched.body.getReader(); @@ -157,6 +241,10 @@ async function watchOnce( if (done) return decoder.end(); lines = decoder.push(value); } catch (error) { + if (error instanceof Truncated) { + // The checkpoint stands: it names the last event applied, and the partial one was not. + return void hooks.error("INTERNAL", `native watch: ${error.message}`, false); + } if (!(error instanceof Malformed)) throw error; return malformed("watch", error); } @@ -165,7 +253,7 @@ async function watchOnce( if (aborted()) return; let more: boolean; try { - more = deliver(line, collection.type, hooks); + more = deliver(line, position.type, hooks, advance, position); } catch (error) { if (!(error instanceof Malformed)) throw error; return malformed("watch", error); @@ -178,7 +266,8 @@ async function watchOnce( await reader.cancel().catch(() => {}); } } catch { - // The network failed. This connection is over, and the lifecycle decides about the next. Host + // The network failed. This connection is over, and the lifecycle decides about the next. The + // checkpoint still names the last event applied, so the next connection resumes from it. Host // callbacks never land here: the lifecycle catches their exceptions itself. } finally { controller.abort(); @@ -186,8 +275,15 @@ async function watchOnce( } } -/** Apply one watch frame; false when the connection must end. Throws Malformed. */ -function deliver(line: string, type: ItemType | undefined, hooks: TransportHooks): boolean { +/** Apply one watch frame, then advance the checkpoint past it; false when the connection must end. + * Throws Malformed. */ +function deliver( + line: string, + type: ItemType | undefined, + hooks: TransportHooks, + advance: (resourceVersion: unknown) => void, + position: Position, +): boolean { let frame: unknown; try { frame = JSON.parse(line); @@ -200,22 +296,28 @@ function deliver(line: string, type: ItemType | undefined, hooks: TransportHooks 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") }); + case "MODIFIED": { + const applied = resource(object, type, "watch"); + hooks.consume({ type: frame.type === "ADDED" ? "added" : "modified", object: applied }); + advance(applied.metadata.resourceVersion); return true; - case "DELETED": - hooks.consume({ type: "deleted", identity: identity(resource(object, type, "watch")) }); + } + case "DELETED": { + const removed = resource(object, type, "watch"); + hooks.consume({ type: "deleted", identity: identity(removed) }); + advance(removed.metadata.resourceVersion); 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. + // object, and the consumer never sees it. It only moves where the next watch resumes. + advance(isRecord(object.metadata) ? object.metadata.resourceVersion : undefined); return true; case "ERROR": { - const { code, terminal, message, retryAfterMs } = statusError(object); + const { code, terminal, message, retryAfterMs, keepsCheckpoint } = statusError(object); + if (!keepsCheckpoint) position.checkpoint = undefined; hooks.error(code, message, terminal, retryAfterMs); - // The API server ends the watch after an error; either way, the next connection re-lists. + // The API server ends the watch after an error. A transient one resumes; any other re-lists. return false; } default: @@ -229,11 +331,13 @@ 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. */ +/** A watch ERROR event's Kubernetes Status, classified like the HTTP status it carries. Only a coded + * transient error keeps the checkpoint: an expiry names the position itself as the problem, and an + * error without a code cannot be told apart from one, so it re-lists. */ 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 expired = reason === "Expired" || reason === "Gone"; + const code = typeof status.code === "number" ? status.code : expired ? 410 : undefined; const { code: errorCode, terminal } = code === undefined ? { code: "INTERNAL" as const, terminal: false } : classify(code); const text = typeof status.message === "string" ? status.message : ""; @@ -244,6 +348,7 @@ function statusError(status: Record) { terminal, message: text !== "" ? text : `native watch: error ${code ?? reason ?? "without a status"}`, retryAfterMs: typeof seconds === "number" && seconds >= 0 ? seconds * 1000 : undefined, + keepsCheckpoint: code !== undefined && !expired && resumable(code), }; } @@ -333,7 +438,8 @@ function identity(object: KRMObject): Identity { /** 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. */ + * complete. Throws Malformed on invalid UTF-8, and Truncated when the stream ends mid-frame — the two + * are kept apart because only malformed input casts doubt on what was already delivered. */ export class WatchDecoder { #utf8 = new TextDecoder("utf-8", { fatal: true }); #buffer = ""; @@ -352,10 +458,16 @@ export class WatchDecoder { return lines; } - /** The stream ended. Anything still buffered is a truncated frame. */ + /** The stream ended. Anything still buffered — a character cut off included — is a truncated frame: + * every complete line has already been returned, so the cut can only be inside the last one. */ end(): void { - this.#buffer += decode(this.#utf8); - if (this.#buffer.trim() !== "") throw new Malformed("the watch ended inside a frame"); + let rest: string; + try { + rest = this.#utf8.decode(); + } catch { + throw new Truncated("the watch ended inside a UTF-8 character"); + } + if ((this.#buffer + rest).trim() !== "") throw new Truncated("the watch ended inside a frame"); } } diff --git a/packages/krm-stream/test/native-editing.test.ts b/packages/krm-stream/test/native-editing.test.ts index c018163..c2d83cc 100644 --- a/packages/krm-stream/test/native-editing.test.ts +++ b/packages/krm-stream/test/native-editing.test.ts @@ -4,7 +4,7 @@ import assert from "node:assert/strict"; import { test } from "node:test"; -import { type KRMObject, LiveResourceStore } from "../src/index.ts"; +import { applyStreamEvent, connectNativeWatch, type KRMObject, LiveResourceStore } from "../src/index.ts"; const { nativeEditor, NativeRequestError } = await import("../../../examples/native-editor/editor.ts"); @@ -358,3 +358,87 @@ test("an accepted write is confirmed by its echo, or by a guarded read when no e ], ); }); + +test("saving is refused while a native watch recovers from expiry, and works again once a resume is live", async () => { + // The watch side: a scripted proxy for LIST and WATCH, driving the real connector into the store. + const encoder = new TextEncoder(); + const open = () => { + let controller!: ReadableStreamDefaultController; + const response = new Response( + new ReadableStream({ + start: (c) => { + controller = c; + }, + }), + ); + return { + response, + push: (text: string) => controller.enqueue(encoder.encode(text)), + end: () => controller.close(), + }; + }; + const frame = (type: string, object: unknown) => `${JSON.stringify({ type, object })}\n`; + const list = (rv: string, ...items: KRMObject[]) => + Response.json({ kind: "ConfigMapList", apiVersion: "v1", metadata: { resourceVersion: rv }, items }); + const watches = [open(), open(), open()]; + let relist!: () => void; + const relisted = new Promise((resolve) => { + relist = resolve; + }); + const watchRequests: string[] = []; + const steps: (() => Response | Promise)[] = [ + () => list("1", native("1")), + () => watches[0]!.response, + // The replacement snapshot after the expiry is held, so the test can save while it is incomplete. + async () => { + await relisted; + return list("3", native("3", "base")); + }, + () => watches[1]!.response, + () => watches[2]!.response, + ]; + const watchFetch = (async (input: RequestInfo | URL) => { + watchRequests.push(String(input)); + return steps[watchRequests.length - 1]!(); + }) as typeof fetch; + const store = new LiveResourceStore(); + const connection = connectNativeWatch( + "/k8s/api/v1/namespaces/app/configmaps?fieldSelector=metadata.name%3Dsettings", + (event) => applyStreamEvent(store, event), + { fetch: watchFetch, retryDelayMs: 0 }, + ); + const states: string[] = []; + connection.subscribe((state) => states.push(state.status)); + const reached = async (lives: number) => { + while (states.filter((s) => s === "live").length < lives) await new Promise((resolve) => setTimeout(resolve, 1)); + }; + const live = () => connection.state.status === "live"; + await reached(1); + + store.setValue("u", ["data", "value"], "mine"); + const { calls, request } = proxy(() => new Response(null, { status: 200 })); + const editor = nativeEditor(store, "u", source, request, live); + + // The history expires: until the replacement snapshot completes, no write is attempted. + watches[0]!.push(frame("ERROR", { kind: "Status", code: 410, reason: "Expired", message: "too old" })); + watches[0]!.end(); + while (watchRequests.length < 3) await new Promise((resolve) => setTimeout(resolve, 1)); + assert.notEqual(connection.state.status, "live"); + assert.equal(await editor.save(), "recovering"); + assert.equal(calls.length, 0); + relist(); + await reached(2); + assert.deepEqual(store.draft("u").data, { value: "mine" }, "the draft survived the re-list"); + assert.equal(store.server("u").metadata.resourceVersion, "3"); + + // A routine EOF resumes from the checkpoint with no snapshot: the draft is untouched, and the save + // goes out once the resumed watch is live, at the version the store holds. + watches[1]!.end(); + await reached(3); + assert.equal(watchRequests.length, 5, "no LIST for the resume"); + assert.equal(new URL(watchRequests[4]!, "http://host").searchParams.get("resourceVersion"), "3"); + assert.equal(await editor.save(), "saved"); + assert.deepEqual(calls[0]!.body, { metadata: { uid: "u", resourceVersion: "3" }, data: { value: "mine" } }); + connection.close(); + await connection.closed; +}); diff --git a/packages/krm-stream/test/native.test.ts b/packages/krm-stream/test/native.test.ts index 580b291..282f1a2 100644 --- a/packages/krm-stream/test/native.test.ts +++ b/packages/krm-stream/test/native.test.ts @@ -91,6 +91,10 @@ function scripted(...steps: Step[]) { return { fetch, urls }; } const isWatch = (url: string) => new URL(url, "http://host").searchParams.get("watch") === "1"; +/** The resourceVersion a WATCH asked to start from. */ +const rvOf = (url: string) => new URL(url, "http://host").searchParams.get("resourceVersion"); +/** Each request as `list` or `watch@`, so a test reads which reconnects resumed. */ +const requested = (urls: string[]) => urls.map((u) => (isWatch(u) ? `watch@${rvOf(u)}` : "list")); /** A step that never answers until the request is aborted, like a slow server. */ const hang: Step = (_url, signal) => new Promise((_resolve, reject) => { @@ -132,7 +136,9 @@ test("an empty collection is a complete snapshot, and the watch opens from its r 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.get("allowWatchBookmarks"), "true", "bookmarks keep a quiet checkpoint fresh"); assert.equal(watch.searchParams.has("limit"), false); + assert.equal(watch.searchParams.has("sendInitialEvents"), false); }); test("requests use same-origin credentials, no cache, cancellation and the caller's headers", async () => { @@ -408,11 +414,9 @@ for (const [label, tail, pattern] of [ /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 () => { + test(`${label} is reported, never skipped, and recovered by a fresh list rather than a resume`, async () => { const errors: unknown[][] = []; const types: string[] = []; const store = new LiveResourceStore(readOnlyPolicy); @@ -505,16 +509,16 @@ test("repeated expiry consumes the bounded budget with backoff instead of loopin assert.equal(handle.state.status, "exhausted"); }); -test("short EOFs on an accepted watch also exhaust the budget", async () => { - let calls = 0; +test("short EOFs on an accepted watch also exhaust the budget, resuming rather than re-listing", async () => { + const requests: string[] = []; const fetch = (async (input: RequestInfo | URL) => { - calls++; + requests.push(isWatch(String(input)) ? "watch" : "list"); 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.deepEqual(requests, ["list", "watch", "watch", "watch"]); assert.equal(states.filter((s) => s === "live").length, 3); assert.equal(handle.state.status, "exhausted"); }); @@ -667,14 +671,16 @@ test("network failures on list, watch and body read consume exactly the retry bu assert.equal(handle.state.status, "exhausted"); }); -test("missed deletes and selector exits are repaired by re-listing; prior state is kept until then", async () => { +test("after expiry, 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. + const { fetch, urls } = scripted( + // The first connection sees a, b and c; its watch then expires without reporting b's deletion + // or c's label change taking it out of the selector, so there is nothing to resume from. () => 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. + () => + body(frames({ type: "ERROR", object: status(410, "too old resource version", { reason: "Expired" }) })).response, + // The replacement snapshot no longer lists them, but its watch is refused: incomplete, so the + // next connection lists again rather than resuming from a snapshot that never completed. () => collection([cm("a", "one", "5")], "5"), () => json(status(503, "busy"), 503), () => collection([cm("a", "one", "6")], "6"), @@ -699,6 +705,553 @@ test("missed deletes and selector exits are repaired by re-listing; prior state await handle.closed; assert.deepEqual(seen, [["a", "b", "c"], ["a", "b", "c"], ["a", "b", "c"], ["a"]]); assert.equal(store.server("a").metadata.resourceVersion, "6"); + assert.deepEqual(requested(urls), ["list", "watch@1", "list", "watch@5", "list", "watch@6"]); +}); + +// ------------------------------------------------------------------------------------ resume -- +// +// After a complete snapshot, a handle resumes its WATCH from the last event its consumer applied (or +// the last bookmark) instead of listing again. A resumed watch starts no snapshot: no LIST, no +// `reset`, no `synced`, and the state goes connecting → live on acceptance. + +const bookmark = (rv: string) => ({ + type: "BOOKMARK", + object: { kind: "ConfigMap", apiVersion: "v1", metadata: { resourceVersion: rv } }, +}); +const expired = (message = "too old resource version") => ({ + type: "ERROR", + object: status(410, message, { reason: "Expired" }), +}); +/** Close the handle the n-th time it goes live. */ +function closeOnLive(handle: ResourceStreamHandle, n: number) { + let lives = 0; + handle.subscribe((state) => { + if (state.status === "live" && ++lives === n) handle.close(); + }); +} + +test("an EOF resumes the watch from the last applied event and replays what was missed, without a snapshot", async () => { + const store = new LiveResourceStore(readOnlyPolicy); + const { fetch, urls } = scripted( + () => + collection([cm("a", "one", "1"), cm("b", "two", "1"), cm("c", "three", "1"), cm("d", "same", "1", "first")], "1"), + // The first watch delivers one change, then ends routinely. + () => body(frames({ type: "MODIFIED", object: cm("a", "one", "2", "seen") })).response, + // Meanwhile a changed, b was deleted, c's labels took it out of the selector and d was replaced + // under the same name. The resumed watch replays all of it, in order. + () => + body( + frames( + { type: "MODIFIED", object: cm("a", "one", "3", "missed") }, + { type: "DELETED", object: cm("b", "two", "4") }, + { type: "DELETED", object: cm("c", "three", "5") }, + { type: "DELETED", object: cm("d", "same", "6", "first") }, + { type: "ADDED", object: cm("d2", "same", "7", "second") }, + ), + { open: true }, + ).response, + ); + const types: string[] = []; + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + if (event.type === "added" && event.object.metadata.uid === "d2") handle.close(); + }, + { fetch, retryDelayMs: 0 }, + ); + const states = statuses(handle); + await handle.closed; + + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@2"], "the reconnect resumed: no second LIST"); + const resumed = new URL(urls[2]!, "http://host"); + assert.equal(resumed.pathname, "/k8s/api/v1/namespaces/app/configmaps"); + assert.equal(resumed.searchParams.get("labelSelector"), "tier=web", "the same selectors"); + assert.equal(resumed.searchParams.get("allowWatchBookmarks"), "true"); + assert.deepEqual(types, [ + ...["reset", "added", "added", "added", "added", "synced", "modified"], + ...["modified", "deleted", "deleted", "deleted", "added"], + ]); + assert.deepEqual(store.ids().sort(), ["a", "d2"]); + assert.deepEqual(store.server("a").data, { value: "missed" }); + assert.equal(store.server("d2").metadata.name, "same"); + assert.deepEqual(states, ["connecting", "syncing", "syncing", "live", "retrying", "connecting", "live", "closed"]); +}); + +test("a resumed watch is live only once accepted, and never syncing", async () => { + let accept!: () => void; + const accepted = new Promise((resolve) => { + accept = resolve; + }); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")], "1"), + () => body("").response, + async () => { + await accepted; + return body("", { open: true }).response; + }, + ); + const types: string[] = []; + const handle = connectNativeWatch(url, (event) => types.push(event.type), { fetch, retryDelayMs: 0 }); + const states = statuses(handle); + closeOnLive(handle, 2); + while (urls.length < 3) await new Promise((resolve) => setTimeout(resolve, 1)); + await new Promise((resolve) => setTimeout(resolve, 10)); + assert.equal(handle.state.status, "connecting", "not live while the resumed watch is opening"); + accept(); + await handle.closed; + assert.deepEqual(states, ["connecting", "syncing", "syncing", "live", "retrying", "connecting", "live", "closed"]); + assert.deepEqual(types, ["reset", "added", "synced"], "acceptance alone delivers nothing"); +}); + +test("a bookmark moves the checkpoint, and changes no object and reaches no consumer", async () => { + const store = new LiveResourceStore(readOnlyPolicy); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")], "1"), + () => body(frames({ type: "MODIFIED", object: cm("a", "one", "2", "changed") }, bookmark("9"))).response, + // A quiet collection: nothing but a bookmark, and the checkpoint still moves. + () => body(frames(bookmark("15"))).response, + () => body("", { open: true }).response, + ); + const types: string[] = []; + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + }, + { fetch, retryDelayMs: 0 }, + ); + closeOnLive(handle, 3); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@9", "watch@15"]); + assert.deepEqual(types, ["reset", "added", "synced", "modified"]); + assert.equal(store.server("a").metadata.resourceVersion, "2", "a bookmark is never an object's version"); + assert.deepEqual(store.server("a").data, { value: "changed" }); +}); + +test("resourceVersions are opaque: the last one in stream order is resumed from, whatever it looks like", async () => { + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "Zm9v/+=")], "Zm9v/+="), + () => + body( + frames( + { type: "MODIFIED", object: cm("a", "one", "10") }, + // Smaller as a number and as a string, and still the latest: never compared. + { type: "MODIFIED", object: cm("a", "one", "9") }, + { type: "ADDED", object: cm("b", "two", "ä b&c=d#e") }, + ), + ).response, + () => body("", { open: true }).response, + ); + const handle = connectNativeWatch(url, ignore, { fetch, retryDelayMs: 0 }); + closeOnLive(handle, 2); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@Zm9v/+=", "watch@ä b&c=d#e"]); +}); + +for (const [label, tail, pattern] of [ + ["a frame", '{"type":"MODIFIED","object":{"metad', /ended inside a frame/], + ["a character", new Uint8Array([0xe2, 0x9c]), /ended inside a UTF-8 character/], +] as const) + test(`a watch that ends inside ${label} resumes from the last complete event`, async () => { + const errors: unknown[][] = []; + const store = new LiveResourceStore(readOnlyPolicy); + // Frames split every 3 bytes, then cut off inside the next one. + const first = body(frames({ type: "MODIFIED", object: cm("a", "one", "2", "complete ✓") }), { + chunk: 3, + open: true, + }); + first.push(tail); + first.end(); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")], "1"), + () => first.response, + // The resumed watch's frames are split anywhere too, multi-byte characters included. + () => + body(frames({ type: "MODIFIED", object: cm("a", "one", "3", "après 🚀") }), { chunk: 1, open: true }).response, + ); + const types: string[] = []; + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + if (event.type === "modified" && event.object.metadata.resourceVersion === "3") handle.close(); + }, + { fetch, retryDelayMs: 0, onError: (...args) => errors.push(args) }, + ); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@2"], "the partial frame was never applied"); + assert.deepEqual(types, ["reset", "added", "synced", "modified", "modified"]); + assert.deepEqual(store.server("a").data, { value: "après 🚀" }); + assert.equal(errors.length, 1); + assert.equal(errors[0]![0], "INTERNAL"); + assert.match(String(errors[0]![1]), pattern); + assert.equal(errors[0]![2], false); + }); + +for (const [label, frame, applied] of [ + ["a frame that is not JSON", "{not json}\n", []], + ["an in-stream error without a code", frames({ type: "ERROR", object: { kind: "Status", message: "odd" } }), []], + [ + "an event without a resourceVersion", + frames({ type: "MODIFIED", object: { ...cm("a", "one", ""), metadata: { uid: "a", name: "one" } } }), + ["modified"], + ], + ["a bookmark without a resourceVersion", frames({ type: "BOOKMARK", object: { kind: "ConfigMap" } }), []], +] as const) + test(`${label} on a resumed watch discards the checkpoint: the next connection lists`, async () => { + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")], "1"), + () => body("").response, + () => body(frame).response, + () => collection([cm("a", "one", "5")], "5"), + () => body("", { open: true }).response, + ); + const types: string[] = []; + const handle = connectNativeWatch(url, (event) => types.push(event.type), { fetch, retryDelayMs: 0 }); + closeOnLive(handle, 3); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@1", "list", "watch@5"]); + assert.deepEqual(types, ["reset", "added", "synced", ...applied, "reset", "added", "synced"]); + }); + +for (const [label, step] of [ + ["refused with 503", () => json(status(503, "busy"), 503)], + [ + "lost to the network", + () => { + throw new TypeError("offline"); + }, + ], + ["expired with HTTP 410", () => json(status(410, "too old resource version", { reason: "Expired" }), 410)], +] as [string, Step][]) + test(`an initialization whose watch is ${label} leaves no checkpoint: the next connection lists and prunes`, async () => { + const store = new LiveResourceStore(readOnlyPolicy); + store.applyServerEvent(cm("stale", "gone", "0")); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1"), cm("b", "two", "1")], "1"), + step, + () => collection([cm("a", "one", "3")], "3"), + () => body("", { open: true }).response, + ); + const types: string[] = []; + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + }, + { fetch, retryDelayMs: 0 }, + ); + closeOnLive(handle, 1); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@1", "list", "watch@3"]); + assert.deepEqual(types, ["reset", "added", "added", "reset", "added", "synced"]); + assert.deepEqual(store.ids(), ["a"], "the completed snapshot pruned b and what preceded both"); + }); + +test("a consumer exception on a resumed watch closes the stream with it and requests nothing more", async () => { + const boom = new Error("render failed"); + const resumed = body( + frames({ type: "MODIFIED", object: cm("a", "one", "3") }, { type: "MODIFIED", object: cm("a", "one", "4") }), + { open: true }, + ); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")], "1"), + () => body(frames({ type: "MODIFIED", object: cm("a", "one", "2") })).response, + () => resumed.response, + ); + const seen: string[] = []; + const handle = connectNativeWatch( + url, + (event) => { + seen.push(event.type === "modified" ? `modified@${event.object.metadata.resourceVersion}` : event.type); + if (event.type === "modified" && event.object.metadata.resourceVersion === "3") throw boom; + }, + { fetch, retryDelayMs: 0 }, + ); + const states = statuses(handle); + await assert.rejects(handle.closed, (error) => error === boom); + await new Promise((resolve) => setTimeout(resolve, 5)); + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@2"], "never retried"); + assert.deepEqual(seen, ["reset", "added", "synced", "modified@2", "modified@3"], "nothing after the exception"); + assert.equal(resumed.state.cancelled, true); + assert.equal(states.at(-1), "closed"); +}); + +test("HTTP 410 and an in-stream 410 on a resumed watch recover with a fresh, pruning snapshot", async () => { + const errors: unknown[][] = []; + const store = new LiveResourceStore(readOnlyPolicy); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1"), cm("b", "two", "1")], "1"), + () => body(frames({ type: "MODIFIED", object: cm("a", "one", "2") })).response, + () => json(status(410, "too old resource version: 2 (7)", { reason: "Expired" }), 410), + () => collection([cm("a", "one", "7")], "7"), + () => body(frames({ type: "MODIFIED", object: cm("a", "one", "8") })).response, + () => body(frames(expired("too old resource version: 8 (9)"))).response, + () => collection([cm("a", "one", "9"), cm("c", "three", "9")], "9"), + () => body("", { open: true }).response, + ); + const types: string[] = []; + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + }, + { fetch, retryDelayMs: 0, onError: (...args) => errors.push(args) }, + ); + const states = statuses(handle); + closeOnLive(handle, 4); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@2", "list", "watch@7", "watch@8", "list", "watch@9"]); + assert.deepEqual(errors, [ + ["RESYNC_REQUIRED", "too old resource version: 2 (7)", false, undefined], + ["RESYNC_REQUIRED", "too old resource version: 8 (9)", false, undefined], + ]); + assert.deepEqual(types, [ + ...["reset", "added", "added", "synced", "modified"], + ...["reset", "added", "synced", "modified"], + ...["reset", "added", "added", "synced"], + ]); + assert.deepEqual(store.ids().sort(), ["a", "c"], "b was pruned by the completed replacement snapshot"); + // Never live from an expiry until the replacement snapshot completes. + assert.deepEqual(states, [ + ...["connecting", "syncing", "syncing", "live", "retrying"], + ...["connecting", "retrying"], + ...["connecting", "syncing", "syncing", "live", "retrying"], + ...["connecting", "live", "retrying"], + ...["connecting", "syncing", "syncing", "live", "closed"], + ]); +}); + +test("repeated expiry on resume consumes the bounded budget with backoff, never an immediate re-list", async () => { + const log: string[] = []; + let lists = 0; + let initial = 0; + const fetch = (async (input: RequestInfo | URL) => { + const u = String(input); + if (!isWatch(u)) { + log.push("list"); + return collection([], String(++lists)); + } + log.push(`watch@${rvOf(u)}`); + // Every snapshot's watch passes one bookmark and ends; every resume from it has expired. + return rvOf(u)!.startsWith("b") + ? body(frames(expired())).response + : body(frames(bookmark(`b${++initial}`))).response; + }) as typeof globalThis.fetch; + const errors: string[] = []; + const handle = connectNativeWatch(url, ignore, { + fetch, + maxRetries: 3, + retryDelayMs: 1, + maxRetryDelayMs: 4, + onError: (code) => errors.push(code), + }); + handle.subscribe((state) => { + if (state.status === "retrying" || state.status === "exhausted") log.push(state.status); + }); + await handle.closed; + assert.deepEqual(log, [ + ...["list", "watch@1", "retrying"], + ...["watch@b1", "retrying"], + ...["list", "watch@2", "retrying"], + ...["watch@b2", "exhausted"], + ]); + assert.deepEqual(errors, ["RESYNC_REQUIRED", "RESYNC_REQUIRED"]); + assert.equal(handle.state.status, "exhausted"); +}); + +for (const [label, step, code] of [ + ["HTTP 403", () => json(status(403, "access revoked"), 403), "FORBIDDEN"], + ["HTTP 401", () => json(status(401, "access revoked"), 401), "UNAUTHENTICATED"], + [ + "an in-stream 403", + () => body(frames({ type: "ERROR", object: status(403, "access revoked", { reason: "Forbidden" }) })).response, + "FORBIDDEN", + ], +] as [string, Step, string][]) + test(`${label} on a resumed watch is terminal`, async () => { + const errors: unknown[][] = []; + const { fetch, urls } = scripted( + () => collection([], "1"), + () => body("").response, + step, + ); + const handle = connectNativeWatch(url, ignore, { fetch, retryDelayMs: 0, onError: (...args) => errors.push(args) }); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@1"], "never retried or re-listed"); + assert.deepEqual(errors, [[code, "access revoked", true, undefined]]); + assert.equal(handle.state.status, "terminal"); + }); + +test("transient failures on a resumed watch resume again from the same checkpoint", async () => { + const errors: string[] = []; + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1")], "1"), + () => body(frames({ type: "MODIFIED", object: cm("a", "one", "2") })).response, + () => json(status(503, "busy"), 503), + () => json(status(429, "slow down"), 429), + () => + body(frames({ type: "ERROR", object: status(500, "etcd leader changed", { reason: "InternalError" }) })).response, + () => body(frames({ type: "ERROR", object: status(504, "timeout", { reason: "Timeout" }) })).response, + () => { + throw new TypeError("offline"); + }, + () => body("").response, + () => body("", { open: true }).response, + ); + const types: string[] = []; + const handle = connectNativeWatch(url, (event) => types.push(event.type), { + fetch, + retryDelayMs: 0, + onError: (code) => errors.push(code), + }); + closeOnLive(handle, 5); + await handle.closed; + assert.deepEqual(requested(urls), ["list", "watch@1", ...Array(7).fill("watch@2")]); + assert.deepEqual(types, ["reset", "added", "synced", "modified"]); + assert.deepEqual(errors, ["UPSTREAM_UNAVAILABLE", "UPSTREAM_UNAVAILABLE", "INTERNAL", "UPSTREAM_UNAVAILABLE"]); +}); + +test("close during a resumed watch's request, its body and the backoff before it", async (t) => { + for (const at of ["resumed watch fetch", "resumed watch body", "backoff"] as const) { + await t.test(at, async () => { + const before = pendingTimeouts(); + const resumed = body("", { open: true }); + const steps: Step[] = [ + () => collection([cm("a", "one", "1")], "1"), + () => body("").response, + ...(at === "backoff" ? [] : [at === "resumed watch fetch" ? hang : () => resumed.response]), + ]; + const { fetch, urls } = scripted(...steps); + const events: string[] = []; + const handle = connectNativeWatch(url, (event) => events.push(event.type), { + fetch, + retryDelayMs: at === "backoff" ? 60_000 : 0, + healthyResetMs: 60_000, + }); + const states = statuses(handle); + if (at === "backoff") await until(handle, "retrying"); + else if (at === "resumed watch body") { + while (states.filter((s) => s === "live").length < 2) await new Promise((resolve) => setTimeout(resolve, 1)); + resumed.push(frames({ type: "MODIFIED", object: cm("a", "one", "2") })); + } else while (urls.length < 3) await new Promise((resolve) => setTimeout(resolve, 1)); + await new Promise((resolve) => setTimeout(resolve, 5)); + handle.close(); + await handle.closed; + const count = events.length; + 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(urls.length, steps.length, "nothing is requested after close"); + assert.deepEqual(events, ["reset", "added", "synced", ...(at === "resumed watch body" ? ["modified"] : [])]); + assert.equal(requested(urls).filter((r) => r === "list").length, 1, "the reconnect never listed"); + if (at === "resumed watch body") assert.equal(resumed.state.cancelled, true); + if (at === "resumed watch fetch") assert.equal(states.filter((s) => s === "live").length, 1); + assert.equal(pendingTimeouts(), before, "the backoff and health timers are released"); + }); + } +}); + +test("each handle keeps its own checkpoint, even on the same URL, and a new handle starts with a LIST", async () => { + const requests: Record = { a: [], b: [], c: [] }; + const scripts: Record = { + a: [ + () => collection([cm("x", "one", "1")], "1"), + () => body(frames({ type: "MODIFIED", object: cm("x", "one", "11") })).response, + () => body("", { open: true }).response, + ], + b: [ + () => collection([cm("x", "one", "1")], "1"), + () => body(frames({ type: "MODIFIED", object: cm("x", "one", "21") })).response, + () => body("", { open: true }).response, + ], + c: [() => collection([cm("x", "one", "30")], "30"), () => body("", { open: true }).response], + }; + // One fetch for every handle: the caller's header tells them apart, as a login identity would. + const fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const who = (init!.headers as Record)["X-Handle"]!; + requests[who]!.push(String(input)); + return scripts[who]![requests[who]!.length - 1]!(String(input), init!.signal!); + }) as typeof globalThis.fetch; + const a = connectNativeWatch(url, ignore, { fetch, retryDelayMs: 0, headers: { "X-Handle": "a" } }); + const b = connectNativeWatch(url, ignore, { fetch, retryDelayMs: 0, headers: { "X-Handle": "b" } }); + closeOnLive(a, 2); + closeOnLive(b, 2); + await Promise.all([a.closed, b.closed]); + const c = connectNativeWatch(url, ignore, { fetch, retryDelayMs: 0, headers: { "X-Handle": "c" } }); + closeOnLive(c, 1); + await c.closed; + assert.deepEqual(requested(requests.a!), ["list", "watch@1", "watch@11"]); + assert.deepEqual(requested(requests.b!), ["list", "watch@1", "watch@21"]); + assert.deepEqual(requested(requests.c!), ["list", "watch@30"], "a new handle never inherits a checkpoint"); +}); + +test("drafts and conflicts survive a resumed watch, and a surviving draft survives the re-list after expiry", async () => { + const store = new LiveResourceStore(); + const first = body("", { open: true }); + const { fetch, urls } = scripted( + () => collection([cm("a", "one", "1", "base"), cm("b", "two", "1", "base")], "1"), + () => first.response, + // While disconnected, another writer changed a's edited field and an unrelated field of b. + () => + body( + frames( + { type: "MODIFIED", object: cm("a", "one", "5", "theirs") }, + { type: "MODIFIED", object: { ...cm("b", "two", "6", "base"), data: { value: "base", other: "x" } } }, + expired(), + ), + ).response, + // a was deleted after the history expired; b survives, with its unrelated change. + () => collection([{ ...cm("b", "two", "7", "base"), data: { value: "base", other: "x" } }], "7"), + () => body("", { open: true }).response, + ); + const types: string[] = []; + let afterResume: { conflicts: unknown; dirtyA: boolean; patchB: unknown } | undefined; + const handle = connectNativeWatch( + url, + (event) => { + types.push(event.type); + applyStreamEvent(store, event); + }, + { fetch, retryDelayMs: 0 }, + ); + handle.subscribe((state) => { + if (state.status === "retrying" && urls.length === 3) { + afterResume = { + conflicts: store.conflicts("a"), + dirtyA: store.isDirty("a", ["data", "value"]), + patchB: store.patch("b"), + }; + } + }); + closeOnLive(handle, 3); + await until(handle, "live"); + store.setValue("a", ["data", "value"], "mine"); + store.setValue("b", ["data", "value"], "mine too"); + first.end(); + await handle.closed; + + assert.deepEqual(requested(urls), ["list", "watch@1", "watch@1", "list", "watch@7"]); + assert.deepEqual(types, [ + ...["reset", "added", "added", "synced"], + ...["modified", "modified"], + ...["reset", "added", "synced"], + ]); + assert.deepEqual(afterResume, { + conflicts: [{ path: ["data", "value"], theirs: "theirs" }], + dirtyA: true, + patchB: { data: { value: "mine too" } }, + }); + assert.deepEqual(store.ids(), ["b"]); + assert.equal(store.server("b").metadata.resourceVersion, "7"); + assert.deepEqual(store.draft("b").data, { value: "mine too", other: "x" }); + assert.deepEqual(store.patch("b"), { data: { value: "mine too" } }); }); test("close during the list request, the list body, the watch request, the watch body and backoff", async (t) => { @@ -872,7 +1425,15 @@ test("a subscriber or onError that throws stops the stream without retrying", as }); 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"]) { + for (const param of [ + "watch=1", + "resourceVersion=5", + "limit=10", + "continue=x", + "resourceVersionMatch=Exact", + "sendInitialEvents=true", + "allowWatchBookmarks=false", + ]) { assert.throws(() => connectNativeWatch(`/k8s/api/v1/configmaps?${param}`, ignore), /must not set/, param); } });