diff --git a/README.md b/README.md index 72bde17..265fdc1 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,7 @@ await client.verifyApiToken(); ### Talking to Servers With and Without the v2 API -EventSourcingDB offers its v2 API since version 1.3, while older versions offer only the v1 API. This client SDK works with both. Creating a client does not send a request. Instead, the first request of a client asks the server whether it offers the v2 API, by sending a `GET` request to `/api/v2/system/ping`, and the client uses the v2 API if it does, and the v1 API otherwise. The client keeps the answer for its lifetime, so every later request goes straight to the API it found. If the client cannot find out, for example because the server is not reachable, the first request fails with that error, and the next request asks again. +EventSourcingDB offers its v2 API since version 1.3, while older versions offer only the v1 API. This client SDK works with both. Creating a client does not send a request. Instead, the first request of a client asks the server whether it offers the v2 API, by sending a `GET` request to `/api/v2/system/ping`, and the client uses the v2 API if it does, and the v1 API otherwise. The client keeps the answer, so every later request goes straight to the API it found. If the server answers a path of the v2 API with 404 or 405, for example after it was rolled back to a version without it, the client asks again on the next request. If the client cannot find out, for example because the server is not reachable, the first request fails with that error, and the next request asks again. Through the v2 API, all functions that read use the HTTP method `QUERY` ([RFC 10008](https://www.rfc-editor.org/rfc/rfc10008)), which, unlike `GET`, carries a request body. If there is a proxy or a load balancer in front of EventSourcingDB, it has to let `QUERY` requests through. There is no fallback to `POST`. diff --git a/src/Client.ts b/src/Client.ts index abe735d..0fc8126 100644 --- a/src/Client.ts +++ b/src/Client.ts @@ -58,6 +58,8 @@ class Client { // a client finds out which API the server offers, and the client keeps the // answer. Requests that start meanwhile wait for the same answer. If the // client fails to find out, the request fails, and the next one tries again. + // The same holds once the client forgets the answer, as #createV2Transport + // describes. async #getTransport(signal?: AbortSignal): Promise { if (this.#transport !== undefined) { return this.#transport; @@ -68,9 +70,7 @@ class Client { if (this.#detection === undefined) { const detection = detectApiVersion(this.#transportOptions).then(apiVersion => { this.#transport = - apiVersion === 'v2' - ? new V2Transport(this.#transportOptions) - : new V1Transport(this.#transportOptions); + apiVersion === 'v2' ? this.#createV2Transport() : new V1Transport(this.#transportOptions); return this.#transport; }); @@ -88,6 +88,27 @@ class Client { return await waitUnlessAborted(this.#detection, signal); } + // A server that answers a path of the v2 API with 404 or 405 no longer + // offers it, for example because it was rolled back to a version without + // it, so the client forgets the answer, and the next request finds out + // again, while the request itself fails as before. Only the transport that + // the client still keeps makes it forget. So a late answer to a request + // through a transport that the client has forgotten already neither + // disturbs finding out anew, which may run meanwhile, nor discards what + // that finds. + #createV2Transport(): V2Transport { + const transport = new V2Transport({ + ...this.#transportOptions, + forgetApiVersion: (): void => { + if (this.#transport === transport) { + this.#transport = undefined; + } + }, + }); + + return transport; + } + public async ping(signal?: AbortSignal): Promise { const transport = await this.#getTransport(signal); await transport.ping(signal); diff --git a/src/detectApiVersion.ts b/src/detectApiVersion.ts index c2737bc..c2e131a 100644 --- a/src/detectApiVersion.ts +++ b/src/detectApiVersion.ts @@ -4,6 +4,7 @@ import { readResponseText } from './stall/readResponseText.js'; import { runWithStallGuard } from './stall/runWithStallGuard.js'; import { hasShapeOf } from './types/hasShapeOf.js'; import { isString } from './types/isString.js'; +import { isV2NotOffered } from './v2/isV2NotOffered.js'; type ApiVersion = 'v1' | 'v2'; @@ -23,9 +24,6 @@ const parseJson = (text: string): unknown => { } }; -const statusNotFound = 404; -const statusMethodNotAllowed = 405; - // Finds out whether the server offers the v2 API, by pinging it there. A // server that offers only the v1 API, as EventSourcingDB 1.2.0 does, answers // a path of the v2 API with the status 404 or 405. Every other answer than @@ -42,7 +40,7 @@ const detectApiVersion = async ({ method: 'GET', }); - if (response.status === statusNotFound || response.status === statusMethodNotAllowed) { + if (isV2NotOffered(response)) { await readResponseText(response, stallGuard); return 'v1'; } diff --git a/src/rollback.test.ts b/src/rollback.test.ts new file mode 100644 index 0000000..f429cdc --- /dev/null +++ b/src/rollback.test.ts @@ -0,0 +1,408 @@ +import assert from 'node:assert/strict'; +import type { ServerResponse } from 'node:http'; +import { afterEach, beforeEach, suite, test } from 'node:test'; +import { Client } from './Client.js'; +import { DbApiError } from './DbApiError.js'; +import type { ApiVersion } from './detectApiVersion.js'; +import { OutcomeUnknownError } from './OutcomeUnknownError.js'; +import { type Answer, answers, lines } from './testing/answers.js'; +import { answerWithText, type FakeRequest, FakeServer, writeHead } from './testing/FakeServer.js'; + +const json = 'application/json'; +const ndJson = 'application/x-ndjson'; + +const isProbe = (request: FakeRequest): boolean => + request.method === 'GET' && request.path === '/api/v2/system/ping'; + +const getApiVersion = (request: FakeRequest): ApiVersion => + request.path.startsWith('/api/v2/') ? 'v2' : 'v1'; + +const consume = async (iterator: AsyncGenerator): Promise => { + const items: unknown[] = []; + for await (const item of iterator) { + items.push(item); + } + return items; +}; + +// Answers as a server without the v2 API answers a path of it. +const answerAsNotOffered = (response: ServerResponse, status: number): void => { + answerWithText(response, status, status === 404 ? '404 page not found' : 'Method Not Allowed'); +}; + +const answerWith = (response: ServerResponse, { contentType, body }: Answer): void => { + writeHead(response, 200, contentType); + response.end(body); +}; + +interface Call { + name: string; + run: (client: Client) => Promise; + paths: Record; + answers: Record; + isWrite: boolean; +} + +const readEventType: Call = { + name: 'readEventType', + run: async (client: Client): Promise => + await client.readEventType('io.eventsourcingdb.test'), + paths: { v1: '/api/v1/read-event-type', v2: '/api/v2/data/read-event-type' }, + answers: { v1: answers.v1.readEventType, v2: answers.v2.readEventType }, + isWrite: false, +}; + +// A single answer, streams with and without an end, a write, and the +// registration of an event schema, each of which reads its answer differently. +const calls: Call[] = [ + readEventType, + { + name: 'readEvents', + run: async (client: Client): Promise => + await consume(client.readEvents('/', { recursive: true })), + paths: { v1: '/api/v1/read-events', v2: '/api/v2/data/read-events' }, + answers: { + v1: { contentType: ndJson, body: lines.v1.event('0') }, + v2: { contentType: ndJson, body: `${lines.v2.event('0')}${lines.v2.end}` }, + }, + isWrite: false, + }, + { + name: 'observeEvents', + run: async (client: Client): Promise => { + // Observing never ends on its own, so the call stops after the first + // event. + for await (const event of client.observeEvents('/', { recursive: true })) { + return event.id; + } + + throw new Error('Failed to observe an event.'); + }, + paths: { v1: '/api/v1/observe-events', v2: '/api/v2/data/observe-events' }, + answers: { + v1: { contentType: ndJson, body: lines.v1.event('0') }, + v2: { contentType: ndJson, body: lines.v2.event('0') }, + }, + isWrite: false, + }, + { + name: 'writeEvents', + run: async (client: Client): Promise => + await client.writeEvents([ + { + source: 'https://www.eventsourcingdb.io', + subject: '/test', + type: 'io.eventsourcingdb.test', + data: { value: 23 }, + }, + ]), + paths: { v1: '/api/v1/write-events', v2: '/api/v2/data/write-events' }, + answers: { v1: answers.v1.writeEvents, v2: answers.v2.writeEvents }, + isWrite: true, + }, + { + name: 'registerEventSchema', + run: async (client: Client): Promise => + await client.registerEventSchema('io.eventsourcingdb.test', { type: 'object' }), + paths: { v1: '/api/v1/register-event-schema', v2: '/api/v2/data/register-event-schema' }, + answers: { + // Registering an event schema through the v1 API does not read the + // answer. + v1: { contentType: json, body: '{}' }, + v2: answers.v2.registerEventSchema ?? { contentType: ndJson, body: '' }, + }, + isWrite: true, + }, +]; + +const isDbApiErrorWithStatus = + (status: number) => + (error: unknown): boolean => + error instanceof DbApiError && error.status === status; + +// A call fails as before with the error of the server, which a write wraps in +// an OutcomeUnknownError, since the v2 API does not refuse a write with 404 or +// 405. +const isFailureOf = + ({ isWrite }: Call, status: number) => + (error: unknown): boolean => { + if (!isWrite) { + return isDbApiErrorWithStatus(status)(error); + } + + return error instanceof OutcomeUnknownError && isDbApiErrorWithStatus(status)(error.cause); + }; + +// A call fails with the error of the server, which a write wraps in an +// OutcomeUnknownError, unless the server refused it. +const isFailureWithStatus = + (status: number) => + (error: unknown): boolean => + isDbApiErrorWithStatus(status)(error instanceof OutcomeUnknownError ? error.cause : error); + +interface KeptAnswer { + name: string; + status: number; + answer: (call: Call, response: ServerResponse) => void; +} + +// The answers with which a call through the v2 API fails, although the server +// still offers the v2 API. +const keptAnswers: KeptAnswer[] = [ + { + name: 'an error item of the v2 API, which comes with the status 200', + status: 200, + answer: (call: Call, response: ServerResponse): void => { + answerWith(response, { + contentType: call.answers.v2.contentType, + body: lines.v2.error('the store is read-only'), + }); + }, + }, + ...[409, 500, 503].map( + (status): KeptAnswer => ({ + name: `the status ${status} of the v2 API`, + status, + answer: (_call: Call, response: ServerResponse): void => { + answerWithText(response, status, 'failed'); + }, + }), + ), +]; + +suite('a server that is rolled back to a version without the v2 API', { timeout: 30_000 }, () => { + let server: FakeServer; + let client: Client; + let offersV2: boolean; + + const getPaths = (): string[] => + server.requests.filter(request => !isProbe(request)).map(request => request.path); + + // The server answers the first probe itself, as a server that offers the v2 + // API. Once the test rolls it back, it answers every path of the v2 API as a + // server without it. + const respondTo = + (call: Call, notOfferedStatus = 404) => + (request: FakeRequest, response: ServerResponse): void => { + if (isProbe(request)) { + answerAsNotOffered(response, 404); + return; + } + + const apiVersion = getApiVersion(request); + if (apiVersion === 'v2' && !offersV2) { + answerAsNotOffered(response, notOfferedStatus); + return; + } + + answerWith(response, call.answers[apiVersion]); + }; + + beforeEach(async () => { + server = new FakeServer('v2'); + offersV2 = true; + await server.start(); + + client = new Client(server.url, 'secret'); + }); + + afterEach(async () => { + await server.stop(); + }); + + const defineForgetTest = (call: Call, status: number): void => { + test(`fails once with the status ${status} of the v2 API, and finds out again which API the server offers with the next request.`, async (): Promise => { + server.respond = respondTo(call, status); + + await call.run(client); + offersV2 = false; + + await assert.rejects(call.run(client), isFailureOf(call, status)); + assert.equal(server.probeCount, 1); + + await call.run(client); + assert.equal(server.probeCount, 2); + assert.deepEqual(getPaths(), [call.paths.v2, call.paths.v2, call.paths.v1]); + }); + }; + + const defineKeepTest = (call: Call, { name, status, answer }: KeptAnswer): void => { + test(`keeps the API after ${name}.`, async (): Promise => { + server.respond = respondTo(call); + await call.run(client); + + server.respond = (_request: FakeRequest, response: ServerResponse): void => { + answer(call, response); + }; + await assert.rejects(call.run(client), isFailureWithStatus(status)); + + server.respond = respondTo(call); + await call.run(client); + assert.equal(server.probeCount, 1); + + // The same client forgets the API once the server is rolled back. + offersV2 = false; + await assert.rejects(call.run(client), isFailureOf(call, 404)); + await call.run(client); + assert.equal(server.probeCount, 2); + assert.deepEqual(getPaths(), [ + call.paths.v2, + call.paths.v2, + call.paths.v2, + call.paths.v2, + call.paths.v1, + ]); + }); + }; + + for (const call of calls) { + suite(call.name, () => { + for (const status of [404, 405]) { + defineForgetTest(call, status); + } + + for (const keptAnswer of keptAnswers) { + defineKeepTest(call, keptAnswer); + } + }); + } + + test('keeps the API after the status 404 of the v1 API, which it uses once the server is rolled back.', async (): Promise => { + server.respond = respondTo(readEventType); + await readEventType.run(client); + offersV2 = false; + await assert.rejects(readEventType.run(client), isDbApiErrorWithStatus(404)); + await readEventType.run(client); + + // Through the v1 API, 404 is an error such as an unknown event type. + server.respond = (_request: FakeRequest, response: ServerResponse): void => { + answerWithText(response, 404, 'event type not found'); + }; + await assert.rejects( + readEventType.run(client), + (error: unknown): boolean => + error instanceof DbApiError && + error.status === 404 && + error.description === 'event type not found', + ); + + server.respond = respondTo(readEventType); + await readEventType.run(client); + assert.equal(server.probeCount, 2); + assert.deepEqual(getPaths(), [ + readEventType.paths.v2, + readEventType.paths.v2, + readEventType.paths.v1, + readEventType.paths.v1, + readEventType.paths.v1, + ]); + }); + + suite('while it finds out again which API the server offers', () => { + let heldRequests: PromiseWithResolvers[]; + let heldProbe: PromiseWithResolvers; + + // The server holds the requests of the v2 API once it is rolled back, and + // the next probe, so that the test answers them in an order of its own. + // Every probe after that one finds the v1 API at once, so a probe too many + // shows in the number of probes. + beforeEach(async () => { + heldRequests = []; + heldProbe = Promise.withResolvers(); + let isProbeHeld = false; + const respond = respondTo(readEventType); + server.respond = (request: FakeRequest, response: ServerResponse): void => { + if (isProbe(request)) { + if (isProbeHeld) { + respond(request, response); + return; + } + + isProbeHeld = true; + heldProbe.resolve(response); + return; + } + if (getApiVersion(request) === 'v2' && !offersV2) { + heldRequests.shift()?.resolve(response); + return; + } + + respond(request, response); + }; + + await readEventType.run(client); + offersV2 = false; + }); + + // Each request has to arrive before the next one starts, so that the test + // knows which answer belongs to which request. + const sendHeld = async (): Promise<[Promise, ServerResponse]> => { + const held = Promise.withResolvers(); + heldRequests.push(held); + const result = readEventType.run(client); + + return [result, await held.promise]; + }; + + test('shares the probe, and keeps its answer, if requests of the v2 API that started before fail with 404 meanwhile.', async (): Promise => { + const [failsFirst, firstResponse] = await sendHeld(); + const [failsDuringProbe, duringProbeResponse] = await sendHeld(); + const [failsAfterProbe, afterProbeResponse] = await sendHeld(); + + answerAsNotOffered(firstResponse, 404); + await assert.rejects(failsFirst, isDbApiErrorWithStatus(404)); + + const waitingBefore = readEventType.run(client); + const probeResponse = await heldProbe.promise; + + answerAsNotOffered(duringProbeResponse, 404); + await assert.rejects(failsDuringProbe, isDbApiErrorWithStatus(404)); + + const waitingAfter = readEventType.run(client); + answerAsNotOffered(probeResponse, 404); + await Promise.all([waitingBefore, waitingAfter]); + + answerAsNotOffered(afterProbeResponse, 404); + await assert.rejects(failsAfterProbe, isDbApiErrorWithStatus(404)); + + await readEventType.run(client); + assert.equal(server.probeCount, 2); + assert.deepEqual(getPaths(), [ + readEventType.paths.v2, + readEventType.paths.v2, + readEventType.paths.v2, + readEventType.paths.v2, + readEventType.paths.v1, + readEventType.paths.v1, + readEventType.paths.v1, + ]); + }); + + test('keeps the answer of the probe if it finds the v2 API again, and a request of the v2 API that started before fails with 404 afterwards.', async (): Promise => { + const [failsFirst, firstResponse] = await sendHeld(); + const [failsLate, lateResponse] = await sendHeld(); + + answerAsNotOffered(firstResponse, 404); + await assert.rejects(failsFirst, isDbApiErrorWithStatus(404)); + + const waiting = readEventType.run(client); + const probeResponse = await heldProbe.promise; + offersV2 = true; + answerWith(probeResponse, answers.v2.ping); + await waiting; + + answerAsNotOffered(lateResponse, 404); + await assert.rejects(failsLate, isDbApiErrorWithStatus(404)); + + await readEventType.run(client); + assert.equal(server.probeCount, 2); + assert.deepEqual(getPaths(), [ + readEventType.paths.v2, + readEventType.paths.v2, + readEventType.paths.v2, + readEventType.paths.v2, + readEventType.paths.v2, + ]); + }); + }); +}); diff --git a/src/v2/V2Transport.ts b/src/v2/V2Transport.ts index b0c15fc..43009a4 100644 --- a/src/v2/V2Transport.ts +++ b/src/v2/V2Transport.ts @@ -43,11 +43,20 @@ const isEventOfType = (item: V2Item, eventType: string): boolean => const readWriteAnswer = async ( response: Response, stallGuard: StallGuard, - { signal, action }: { signal?: AbortSignal; action: string }, + { + signal, + action, + forgetApiVersion, + }: { signal?: AbortSignal; action: string; forgetApiVersion: () => void }, ): Promise => { const items: V2Item[] = []; - for await (const item of readV2Stream(response, stallGuard, { signal, action, hasEnd: true })) { + for await (const item of readV2Stream(response, stallGuard, { + signal, + action, + hasEnd: true, + forgetApiVersion, + })) { if (item.type === 'end') { return items; } @@ -59,21 +68,29 @@ const readWriteAnswer = async ( throw createStreamEndedError(action); }; +interface V2TransportOptions extends TransportOptions { + // Tells the client to forget which API the server offers, once the server + // answers as one that does not offer the v2 API. + forgetApiVersion: () => void; +} + // Sends the requests through the v2 API, which EventSourcingDB offers since // version 1.3. class V2Transport implements Transport { readonly #url: URL; readonly #apiToken: string; readonly #stallTimeout: number; + readonly #forgetApiVersion: () => void; #getUrl(path: string): string { return new URL(path, this.#url).toString(); } - public constructor({ url, apiToken, stallTimeout }: TransportOptions) { + public constructor({ url, apiToken, stallTimeout, forgetApiVersion }: V2TransportOptions) { this.#url = url; this.#apiToken = apiToken; this.#stallTimeout = stallTimeout; + this.#forgetApiVersion = forgetApiVersion; } public async ping(signal?: AbortSignal): Promise { @@ -85,7 +102,10 @@ class V2Transport implements Transport { method: 'GET', }); - const item = await readV2Answer(response, stallGuard, 'ping'); + const item = await readV2Answer(response, stallGuard, { + action: 'ping', + forgetApiVersion: this.#forgetApiVersion, + }); if (!isEventOfType(item, 'io.eventsourcingdb.api.ping-received')) { throw new Error('Failed to ping.'); } @@ -104,7 +124,10 @@ class V2Transport implements Transport { }, }); - const item = await readV2Answer(response, stallGuard, 'verify API token'); + const item = await readV2Answer(response, stallGuard, { + action: 'verify API token', + forgetApiVersion: this.#forgetApiVersion, + }); if (!isEventOfType(item, 'io.eventsourcingdb.api.token-verified')) { throw new Error('Failed to verify API token.'); } @@ -134,7 +157,11 @@ class V2Transport implements Transport { body, }); - const items = await readWriteAnswer(response, stallGuard, { signal, action }); + const items = await readWriteAnswer(response, stallGuard, { + signal, + action, + forgetApiVersion: this.#forgetApiVersion, + }); return items.map(item => { if (item.type !== 'event') { @@ -155,6 +182,7 @@ class V2Transport implements Transport { const url = this.#getUrl('/api/v2/data/read-events'); const apiToken = this.#apiToken; const stallTimeout = this.#stallTimeout; + const forgetApiVersion = this.#forgetApiVersion; const action = 'read events'; return streamWithStallGuard({ stallTimeout, signal }, async function* (stallGuard) { @@ -171,6 +199,7 @@ class V2Transport implements Transport { signal, action, hasEnd: true, + forgetApiVersion, })) { if (item.type === 'end') { return; @@ -189,6 +218,7 @@ class V2Transport implements Transport { const url = this.#getUrl('/api/v2/data/run-eventql-query'); const apiToken = this.#apiToken; const stallTimeout = this.#stallTimeout; + const forgetApiVersion = this.#forgetApiVersion; const action = 'run EventQL query'; return streamWithStallGuard({ stallTimeout, signal }, async function* (stallGuard) { @@ -205,6 +235,7 @@ class V2Transport implements Transport { signal, action, hasEnd: true, + forgetApiVersion, })) { if (item.type === 'end') { return; @@ -230,6 +261,7 @@ class V2Transport implements Transport { const url = this.#getUrl('/api/v2/data/observe-events'); const apiToken = this.#apiToken; const stallTimeout = this.#stallTimeout; + const forgetApiVersion = this.#forgetApiVersion; const action = 'observe events'; return streamWithStallGuard({ stallTimeout, signal }, async function* (stallGuard) { @@ -246,6 +278,7 @@ class V2Transport implements Transport { signal, action, hasEnd: false, + forgetApiVersion, })) { if (item.type === 'event') { yield convertV2ItemToEvent(item, action); @@ -283,7 +316,11 @@ class V2Transport implements Transport { body, }); - const items = await readWriteAnswer(response, stallGuard, { signal, action }); + const items = await readWriteAnswer(response, stallGuard, { + signal, + action, + forgetApiVersion: this.#forgetApiVersion, + }); const [item, ...otherItems] = items; if ( @@ -305,6 +342,7 @@ class V2Transport implements Transport { const url = this.#getUrl('/api/v2/data/read-subjects'); const apiToken = this.#apiToken; const stallTimeout = this.#stallTimeout; + const forgetApiVersion = this.#forgetApiVersion; const action = 'read subjects'; return streamWithStallGuard({ stallTimeout, signal }, async function* (stallGuard) { @@ -324,6 +362,7 @@ class V2Transport implements Transport { signal, action, hasEnd: true, + forgetApiVersion, })) { if (item.type === 'end') { return; @@ -355,7 +394,10 @@ class V2Transport implements Transport { }), }); - const item = await readV2Answer(response, stallGuard, action); + const item = await readV2Answer(response, stallGuard, { + action, + forgetApiVersion: this.#forgetApiVersion, + }); if (item.type !== 'eventType') { throw new Error(`Failed to ${action}, got an unexpected answer.`); } @@ -368,6 +410,7 @@ class V2Transport implements Transport { const url = this.#getUrl('/api/v2/data/read-event-types'); const apiToken = this.#apiToken; const stallTimeout = this.#stallTimeout; + const forgetApiVersion = this.#forgetApiVersion; const action = 'read event types'; return streamWithStallGuard({ stallTimeout, signal }, async function* (stallGuard) { @@ -382,6 +425,7 @@ class V2Transport implements Transport { signal, action, hasEnd: true, + forgetApiVersion, })) { if (item.type === 'end') { return; diff --git a/src/v2/isV2NotOffered.ts b/src/v2/isV2NotOffered.ts new file mode 100644 index 0000000..35d608b --- /dev/null +++ b/src/v2/isV2NotOffered.ts @@ -0,0 +1,10 @@ +const statusNotFound = 404; +const statusMethodNotAllowed = 405; + +// Tells whether the server answered a request to a path of the v2 API as a +// server that does not offer the v2 API does, such as EventSourcingDB 1.2.0, +// with the status 404 or 405. +const isV2NotOffered = (response: Response): boolean => + response.status === statusNotFound || response.status === statusMethodNotAllowed; + +export { isV2NotOffered }; diff --git a/src/v2/readV2Answer.ts b/src/v2/readV2Answer.ts index ef449e6..dbe2a71 100644 --- a/src/v2/readV2Answer.ts +++ b/src/v2/readV2Answer.ts @@ -2,15 +2,22 @@ import { assertEventSourcingDb } from '../http/assertEventSourcingDb.js'; import { createDbApiErrorFromResponse } from '../http/createDbApiErrorFromResponse.js'; import { readResponseText } from '../stall/readResponseText.js'; import type { StallGuard } from '../stall/StallGuard.js'; +import { isV2NotOffered } from './isV2NotOffered.js'; import { parseV2Item } from './parseV2Item.js'; import type { V2Item } from './V2Item.js'; -// Reads the single answer of a request to the v2 API. +// Reads the single answer of a request to the v2 API. A server that answers +// with 404 or 405 no longer offers the v2 API, so the client forgets which API +// the server offers, and the request fails as with any other status. const readV2Answer = async ( response: Response, stallGuard: StallGuard, - action: string, + { action, forgetApiVersion }: { action: string; forgetApiVersion: () => void }, ): Promise => { + if (isV2NotOffered(response)) { + forgetApiVersion(); + } + assertEventSourcingDb(response); if (response.status !== 200) { diff --git a/src/v2/readV2Stream.ts b/src/v2/readV2Stream.ts index eae74da..1b79de3 100644 --- a/src/v2/readV2Stream.ts +++ b/src/v2/readV2Stream.ts @@ -3,6 +3,7 @@ import { createDbApiErrorFromResponse } from '../http/createDbApiErrorFromRespon import { createStreamEndedError } from '../http/createStreamEndedError.js'; import { readNdJsonStream } from '../ndjson/readNdJsonStream.js'; import type { StallGuard } from '../stall/StallGuard.js'; +import { isV2NotOffered } from './isV2NotOffered.js'; import { parseV2Item } from './parseV2Item.js'; import type { V2Item } from './V2Item.js'; @@ -10,12 +11,23 @@ import type { V2Item } from './V2Item.js'; // with the error of an error item. A stream that has an end closes with an // end item, which it yields as well, and a stream that has none, such as // observing events, never ends on its own. So a stream that ends otherwise -// broke off and fails, unless the caller aborted it. +// broke off and fails, unless the caller aborted it. A server that answers +// with 404 or 405 no longer offers the v2 API, so the client forgets which API +// the server offers, and the stream fails as with any other status. const readV2Stream = async function* ( response: Response, stallGuard: StallGuard, - { signal, action, hasEnd }: { signal?: AbortSignal; action: string; hasEnd: boolean }, + { + signal, + action, + hasEnd, + forgetApiVersion, + }: { signal?: AbortSignal; action: string; hasEnd: boolean; forgetApiVersion: () => void }, ): AsyncGenerator { + if (isV2NotOffered(response)) { + forgetApiVersion(); + } + assertEventSourcingDb(response); if (response.status !== 200) { diff --git a/src/writeBeforeSending.test.ts b/src/writeBeforeSending.test.ts index 0ed73e9..3ae18e9 100644 --- a/src/writeBeforeSending.test.ts +++ b/src/writeBeforeSending.test.ts @@ -10,10 +10,16 @@ import { apiVersions } from './testing/answers.js'; import { V1Transport } from './v1/V1Transport.js'; import { V2Transport } from './v2/V2Transport.js'; +const keepApiVersion = (): void => { + // Intentionally left blank, since there is no client that could forget. +}; + // The transports are used directly, since the client would first have to find // out which API the server offers, which needs a connection that works. const createTransport = (apiVersion: ApiVersion, options: TransportOptions): Transport => - apiVersion === 'v1' ? new V1Transport(options) : new V2Transport(options); + apiVersion === 'v1' + ? new V1Transport(options) + : new V2Transport({ ...options, forgetApiVersion: keepApiVersion }); interface Write { name: string;