diff --git a/hub/upstream.test.ts b/hub/upstream.test.ts index 75fccd4..2dc6e09 100644 --- a/hub/upstream.test.ts +++ b/hub/upstream.test.ts @@ -33,9 +33,36 @@ describe('Upstream', () => { const up = new Upstream(config, hang, () => now) await expect(up.call(server, 'x')).rejects.toThrow('timed out after 20 ms') await expect(up.call(server, 'y')).rejects.toThrow('timed out') - expect(calls).toBe(1) + // Two: the read and its second try. + expect(calls).toBe(2) now = 11_000 await expect(up.call(server, 'y')).rejects.toThrow('timed out') + expect(calls).toBe(4) + }) + + it('asks a slow read again and takes the answer that comes first', async () => { + let calls = 0 + const up = new Upstream(config, ((_url: string, init: RequestInit) => { + calls++ + if (calls > 1) return Promise.resolve(ok()) + return new Promise((_resolve, reject) => init.signal?.addEventListener('abort', () => reject(Object.assign(new Error('aborted'), { name: 'AbortError' })))) + }) as unknown as typeof fetch) + await expect(up.call(server, 'x')).resolves.toEqual({ n: 1 }) expect(calls).toBe(2) }) + + it('serves the last good answer for two minutes when the server stops answering', async () => { + let now = 0 + let fail = false + const up = new Upstream(config, (async () => { + if (fail) throw new Error('ECONNRESET') + return ok() + }) as typeof fetch, () => now) + await up.call(server, 'x') + fail = true + now = 60_000 + await expect(up.call(server, 'x')).resolves.toEqual({ n: 1 }) + now = 140_000 + await expect(up.call(server, 'x')).rejects.toThrow('ECONNRESET') + }) }) diff --git a/hub/upstream.ts b/hub/upstream.ts index 9971e0e..fb62682 100644 --- a/hub/upstream.ts +++ b/hub/upstream.ts @@ -43,6 +43,12 @@ const WRITE_TIMEOUT_MS = 60_000 /* How long reads of a server that timed out fail at once (the console polls every few seconds). */ const DOWN_MS = 10_000 +/* How long a read may wait before it is asked again alongside. */ +const HEDGE_MS = 2500 + +/* How long past its time a cached answer still stands in for a server that does not answer. */ +const STALE_MS = 120_000 + export class Upstream { #config: HubConfig #fetch: typeof fetch @@ -64,13 +70,18 @@ export class Upstream { const hit = this.#cache.get(key) if (hit !== undefined && hit.until > this.#now()) return hit.response // A server that just timed out fails again at once, rather than holding every read for the whole timeout. + // The last good answer, kept a little past its time: it stands in when the server does not answer. + const stale = hit !== undefined && hit.until + STALE_MS > this.#now() ? hit : undefined const down = this.#down.get(server.id) - if (down !== undefined && down.until > this.#now()) throw down.error + if (down !== undefined && down.until > this.#now()) { + if (stale !== undefined) return stale.response + throw down.error + } // The same read already on its way (the sidebar, the page and a panel ask together) is shared. const pending = this.#pending.get(key) if (pending !== undefined) return pending - const call = this.#exec(server, path, new URLSearchParams(sorted), token, { method: 'GET' }, this.#config.timeoutMs).then( + const call = this.#read(server, path, new URLSearchParams(sorted), token).then( (response) => { this.#down.delete(server.id) if (this.#config.cacheSeconds > 0) { @@ -81,6 +92,7 @@ export class Upstream { }, (e: unknown) => { if (e instanceof UpstreamError && e.message.startsWith('timed out')) this.#down.set(server.id, { until: this.#now() + DOWN_MS, error: e }) + if (stale !== undefined && e instanceof UpstreamError && e.kind === 'unreachable') return stale.response throw e }, ) @@ -92,6 +104,36 @@ export class Upstream { } } + /* + One read, asked a second time when the first is slow to answer: a request can stall on its own + (a lost packet, a TLS handshake that hangs) while the server is fine, and a fresh one then answers + at once. Whichever answers first wins; the first one's error is the error when both fail. + shortcut: the losing request is left to finish or time out by itself, abort it if reads get heavy. + */ + #read(server: HubServer, path: string, query: URLSearchParams, token: string): Promise { + const total = this.#config.timeoutMs + const wait = Math.min(HEDGE_MS, total / 3) + const first = this.#exec(server, path, new URLSearchParams(query), token, { method: 'GET' }, total) + return new Promise((resolve, reject) => { + let second: Promise | undefined + const timer = setTimeout(() => { + second = this.#exec(server, path, new URLSearchParams(query), token, { method: 'GET' }, total - wait) + second.then(resolve, () => {}) + }, wait) + first.then( + (r) => { + clearTimeout(timer) + resolve(r) + }, + (e: unknown) => { + clearTimeout(timer) + if (second === undefined) reject(e) + else second.then(resolve, () => reject(e)) + }, + ) + }) + } + /* A write: POST with the server's ADMIN token and never the read one, never cached. Whatever the outcome, what the hub remembers of this server is dropped afterwards, so the next read