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();