From ddeaaf426b106e19bb7c8807722f781995a3dd48 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 29 Aug 2026 12:34:16 +0000 Subject: [PATCH] fix(adapter-api): stop the feed reconnecting against a refusal that repeats MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit isFatalFeedError ended the reconnect loop only for an unrenewable credential (401) and an authorization refusal (403). Its doc comment already stated the general rule — a response that "will reject the next connection identically, so retrying only spins" — but the predicate implemented a narrower version of it. Every other client-faulting refusal was therefore treated as transient. A GET /changes answered 400 bad_request, which a malformed cursor, kind or include value earns, deserializes to StackQueryError; that is neither class, so subscribeChanges() reported it through onError and retried, saturating at RECONNECT_MAX_MS into an attempt roughly every 15 seconds, indefinitely, sending the same rejected request each time. onReset never fires on that path, so the application had nothing to reconcile from either — it just received the same error on a timer. The predicate now decides on the wire status, which keeps it correct as the error taxonomy grows: a 4xx ends the loop, a 5xx reconnects. The distinction matters most for timeout (503), the answer a server gives while shedding query load — treating that as fatal would turn a busy server into a permanently dead subscription, which is worse than the loop this fixes. Enumerating error classes instead would be one StackTimeoutError away from exactly that, so the status is the safer thing to read. Both directions are pinned by tests: a 400 stops the loop, a 503 does not. The rule is now in docs/spec/wire-format.md § Backpressure and reconnection, which described when a client reconnects but not when it gives up. The status table there already implied it — 503 is documented as "worth retrying — which is why this is not bad_request: that code tells a client its request was malformed and retrying won't help." Closes #199 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FH88yDv3Fk2RmLsU7DU4Rz --- .changeset/olive-donkeys-jam.md | 20 ++++++ docs/spec/wire-format.md | 2 + packages/adapter-api/src/index.ts | 25 +++++--- .../adapter-api/tests/change-feed.test.ts | 63 +++++++++++++++++++ 4 files changed, 100 insertions(+), 10 deletions(-) create mode 100644 .changeset/olive-donkeys-jam.md diff --git a/.changeset/olive-donkeys-jam.md b/.changeset/olive-donkeys-jam.md new file mode 100644 index 0000000..682215e --- /dev/null +++ b/.changeset/olive-donkeys-jam.md @@ -0,0 +1,20 @@ +--- +'@haverstack/adapter-api': patch +--- + +Stop the change-feed client reconnecting against a refusal that will repeat. + +`isFatalFeedError` ended the reconnect loop only for an unrenewable +credential (401) and an authorization refusal (403). Every other refusal the +server faulted the request for — a malformed cursor or filter answered +`400 bad_request`, say — was treated as transient, so `subscribeChanges()` +retried it with backoff indefinitely, settling into an attempt roughly every +15 seconds and reporting the same error to `onError` each time. The +subscriber was never told to stop, and `onReset` never fired, so the +application had nothing to reconcile from either. + +The predicate now decides on the wire status: a `4xx` ends the loop, since +the reconnect sends the same request and would be refused the same way. A +`5xx` still reconnects, which is what keeps `timeout` — the answer a server +gives while shedding query load — from turning a busy server into a +permanently dead subscription. diff --git a/docs/spec/wire-format.md b/docs/spec/wire-format.md index 700ac9a..15f98f1 100644 --- a/docs/spec/wire-format.md +++ b/docs/spec/wire-format.md @@ -539,6 +539,8 @@ data: {"reason":"cursor_expired"} **Reconnection is the client's job, with exponential backoff and jitter.** A server restart otherwise produces a synchronized reconnect stampede from every client it dropped. +**A client stops reconnecting when the answer was `4xx`, and keeps reconnecting when it was `5xx`.** The reconnect sends the same request, so a status faulting that request — a malformed cursor or filter, an unrenewable credential, an authorization refusal — will be answered identically however long the client waits, and backing off only spins. A `5xx` says the server could not serve a request it did not fault, which is the case backoff exists for; `timeout` is the answer a server gives while [shedding query load](#bounding-query-cost) and a client that gave up on it would turn a busy server into a dead subscription. A client that stops reports the error to its subscriber first; the repair is to subscribe again. + ### Permission scoping **A connection delivers the events its token's session may read, and nothing else.** The predicate is literally `canRead` applied per event — no second vocabulary, no feed-specific ACL. A server subscribes **unscoped** at the storage owner and fans out per connection, filtering each through the `ScopedStack` its token's session names via `Stack.forSession()`, taking the `(principalId, subjectId)` pair whole. Delegated authority is then the ordinary [intersection](./access-control.md#delegation-principal-and-subject), inherited rather than reimplemented. diff --git a/packages/adapter-api/src/index.ts b/packages/adapter-api/src/index.ts index 91e4b7f..bea2b9c 100644 --- a/packages/adapter-api/src/index.ts +++ b/packages/adapter-api/src/index.ts @@ -16,7 +16,7 @@ * patchContent()/deleteRecord()/etc.'s expectedVersion option. */ -import { StackQueryError, StackPermissionError } from '@haverstack/core'; +import { StackError, StackQueryError } from '@haverstack/core'; import type { StackAdapter, StackRecord, @@ -59,6 +59,7 @@ import { isRetryableAuthError, isValidSeq, supportsChangeFeed, + WIRE_ERROR_STATUS, supportsDidChallenge, CHANGE_FRAME_READY, CHANGE_FRAME_RECORD, @@ -484,16 +485,20 @@ const reconnectDelay = (attempt: number): number => { const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); /** - * Whether a feed error is one reconnecting cannot recover. An - * authentication failure (a 401 whose re-auth did not renew, or no - * credential to renew with) and an authorization refusal (403) will reject - * the next connection identically, so retrying only spins. A connection or - * server error is transient and reconnects. The stream is closed by - * returning from the pump; the subscriber has already been told via - * onError. + * Whether a feed error is one reconnecting cannot recover. A 4xx faults the + * request, and the reconnect sends the same one, so retrying only spins. A + * 5xx is the server's own trouble and may clear — a shed-load `timeout` + * reconnects. The stream is closed by returning from the pump; the + * subscriber has already been told via onError. */ -const isFatalFeedError = (err: unknown): boolean => - err instanceof APIAdapterAuthError || err instanceof StackPermissionError; +const isFatalFeedError = (err: unknown): boolean => { + if (err instanceof APIAdapterAuthError) return true; + if (err instanceof StackError) { + const status = WIRE_ERROR_STATUS[err.code]; + return status >= 400 && status < 500; + } + return false; +}; // ------------------------------------------------------- // Challenge–response handshake diff --git a/packages/adapter-api/tests/change-feed.test.ts b/packages/adapter-api/tests/change-feed.test.ts index 4ab4cc2..4db6a95 100644 --- a/packages/adapter-api/tests/change-feed.test.ts +++ b/packages/adapter-api/tests/change-feed.test.ts @@ -11,6 +11,7 @@ import { APIAdapterAuthError, } from '../src/index.js'; import { WIRE_PROTOCOL_VERSION } from '@haverstack/wire-types'; +import { StackQueryError, StackTimeoutError } from '@haverstack/core'; import type { RecordChange } from '@haverstack/core'; const BASE_URL = 'https://stack.example.com'; @@ -486,6 +487,68 @@ describe('reconnection', () => { stop(); }); + // A request the server faulted is the same request on the next + // connection, so the loop ends rather than spinning against a verdict + // that will not change. + test('stops reconnecting after a 4xx the reconnect would only repeat', async () => { + vi.useFakeTimers(); + const adapter = await openAdapter(); + const first = feed(); + mockFetch.mockResolvedValueOnce(first.response); + + const onError = vi.fn(); + const subscription = adapter.subscribeChanges({ onError }, () => {}); + first.write(READY); + const stop = await subscription; + + mockFetch.mockResolvedValueOnce( + jsonResponse({ error: { code: 'bad_request', message: 'Invalid cursor' } }, 400), + ); + first.end(); + + await vi.advanceTimersByTimeAsync(60_000); + await vi.waitFor(() => expect(onError).toHaveBeenCalledOnce()); + expect(onError.mock.calls[0]![0]).toBeInstanceOf(StackQueryError); + + // No further reconnect: the fetch count holds at discovery + first + + // the refused reconnect, even after more time passes. + const calls = mockFetch.mock.calls.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(mockFetch.mock.calls.length).toBe(calls); + stop(); + }); + + // 503 is a server shedding load, not a verdict on the request — being + // reconnected against is the whole point of it. + test('keeps reconnecting after a 503 the server may recover from', async () => { + vi.useFakeTimers(); + const adapter = await openAdapter(); + const first = feed(); + mockFetch.mockResolvedValueOnce(first.response); + + const onError = vi.fn(); + const subscription = adapter.subscribeChanges({ onError }, () => {}); + first.write(READY); + const stop = await subscription; + + mockFetch.mockResolvedValueOnce( + jsonResponse({ error: { code: 'timeout', message: 'Shedding load' } }, 503), + ); + const recovered = feed(); + mockFetch.mockResolvedValueOnce(recovered.response); + first.end(); + + await vi.advanceTimersByTimeAsync(60_000); + await vi.waitFor(() => expect(onError).toHaveBeenCalledOnce()); + expect(onError.mock.calls[0]![0]).toBeInstanceOf(StackTimeoutError); + + // discovery + first + the 503 + the attempt that reaches the server + // again: the subscription outlived the refusal. + await vi.advanceTimersByTimeAsync(120_000); + await vi.waitFor(() => expect(mockFetch.mock.calls.length).toBe(4)); + stop(); + }); + test('stops reconnecting once unsubscribed, and aborts the open stream', async () => { vi.useFakeTimers(); const adapter = await openAdapter();