From 83d096064b0f0632ba0c3117fa6884189bcde322 Mon Sep 17 00:00:00 2001 From: Harlan Wilton Date: Tue, 18 Aug 2026 13:25:03 +1000 Subject: [PATCH 1/3] fix(script): report proxy upstream failures as gateway errors Embed and image proxies mirrored the upstream status. A 5xx from Bluesky, Instagram, or a CDN surfaced as a 5xx from the app hosting the proxy, and a transport failure carried no status at all, so it became a 500. Both read as a defect in the host app. They are now 502, or 504 when the request timed out. Nitro stores nothing when a cached resolver throws, so an upstream that keeps refusing a resource (rate limit, login wall, deleted post) was re-fetched on every request. Each attempt raised a server error and the retries deepened the rate limit that caused them. A failed fetch is now replayed for up to 60s before the upstream is tried again. Stale-while-revalidate already covers a resource that succeeded once, so this only gates the cold path. --- .../runtime/server/utils/cached-upstream.ts | 138 +++++++++++++++--- test/unit/cached-upstream-failure.test.ts | 132 +++++++++++++++++ test/unit/cached-upstream.test.ts | 61 +++++++- 3 files changed, 306 insertions(+), 25 deletions(-) create mode 100644 test/unit/cached-upstream-failure.test.ts diff --git a/packages/script/src/runtime/server/utils/cached-upstream.ts b/packages/script/src/runtime/server/utils/cached-upstream.ts index 228cdc13..997a10e2 100644 --- a/packages/script/src/runtime/server/utils/cached-upstream.ts +++ b/packages/script/src/runtime/server/utils/cached-upstream.ts @@ -88,6 +88,11 @@ interface BoundedUpstreamResponse { const DEFAULT_BINARY_MAX_RESPONSE_BYTES = 10 * 1024 * 1024 const DEFAULT_JSON_MAX_RESPONSE_BYTES = 2 * 1024 * 1024 +const TIMEOUT_ERROR_NAMES = new Set(['AbortError', 'BodyTimeoutError', 'HeadersTimeoutError', 'TimeoutError']) +const MAX_TRACKED_FAILURES = 512 + +/** How long a failed upstream fetch is replayed before the upstream is tried again. */ +export const UPSTREAM_FAILURE_MAX_AGE = 60 export function isSafeHttpsUrl(url: URL): boolean { return url.protocol === 'https:' @@ -104,6 +109,71 @@ function upstreamError(message: string, statusCode: number, statusMessage: strin }) } +/** + * Transport failures (DNS, reset connection, timeout) carry no status, so they + * would surface as a 500 and read as a defect in the app hosting the proxy. + */ +function asUpstreamError(error: unknown): Error { + if (typeof (error as { statusCode?: unknown } | null)?.statusCode === 'number') + return error as Error + const timedOut = isTimeoutError(error) + return upstreamError( + `Upstream request failed: ${(error as Error | null)?.message || 'unknown error'}`, + timedOut ? 504 : 502, + timedOut ? 'Gateway Timeout' : 'Upstream request failed', + error, + ) +} + +function isTimeoutError(error: unknown): boolean { + const candidate = error as { name?: string, cause?: { name?: string, code?: string } } | null + const code = candidate?.cause?.code + return TIMEOUT_ERROR_NAMES.has(candidate?.name || '') + || TIMEOUT_ERROR_NAMES.has(candidate?.cause?.name || '') + || (typeof code === 'string' && code.includes('TIMEOUT')) +} + +/** + * Short-lived replay of the last failure for a cache key. + * + * Nitro stores nothing when the resolver throws, so an upstream that keeps + * refusing a resource (rate limit, login wall, deleted post) is re-fetched on + * every request. Each attempt raises a server error, and the retries deepen the + * rate limit that caused them. Stale-while-revalidate already covers a resource + * that was fetched successfully once, so this only gates the cold path. + */ +function createFailureGate(maxAge: number) { + const failures = new Map() + const failureWindow = Math.min(UPSTREAM_FAILURE_MAX_AGE, maxAge) * 1000 + + return { + replay(key: string): void { + const failure = failures.get(key) + if (!failure) + return + if (Date.now() >= failure.until) { + failures.delete(key) + return + } + throw upstreamError(failure.error.message, failure.error.statusCode, failure.error.statusMessage) + }, + record(key: string, error: unknown): void { + const failure = error as Error & { statusCode?: number, statusMessage?: string } + // Insertion order is eviction order; the oldest key is the least useful. + if (failures.size >= MAX_TRACKED_FAILURES) + failures.delete(failures.keys().next().value!) + failures.set(key, { + until: Date.now() + failureWindow, + error: { + message: failure?.message || 'Upstream request failed', + statusCode: failure?.statusCode ?? 502, + statusMessage: failure?.statusMessage || 'Upstream request failed', + }, + }) + }, + } +} + function resolveMaxResponseBytes(value: number | undefined, fallback: number): number { const maxBytes = value ?? fallback if (!Number.isSafeInteger(maxBytes) || maxBytes < 0) @@ -264,10 +334,13 @@ async function fetchBoundedUpstream( } if (!options.ignoreResponseError && response.status >= 400 && response.status < 600) { + // An upstream 5xx is the upstream's fault, not ours. Mirroring it would + // report the app hosting this proxy as broken, so it becomes a 502. + const upstreamFault = response.status >= 500 await rejectResponse(response, upstreamError( `Upstream request failed with status ${response.status}`, - response.status, - response.statusText || 'Upstream request failed', + upstreamFault ? 502 : response.status, + upstreamFault ? 'Upstream request failed' : (response.statusText || 'Upstream request failed'), )) } @@ -298,8 +371,8 @@ async function fetchBoundedUpstream( } } catch (error) { - primaryError = error - throw error + primaryError = asUpstreamError(error) + throw primaryError } finally { await closePublicNetworkDispatcher(network, primaryError) @@ -316,6 +389,24 @@ export function createCachedBinaryFetch( config: CachedBinaryFetchConfig = {}, ): (url: string, opts?: CachedBinaryFetchOptions) => Promise { const maxResponseBytes = resolveMaxResponseBytes(config.maxResponseBytes, DEFAULT_BINARY_MAX_RESPONSE_BYTES) + const failureGate = createFailureGate(maxAge) + const cacheKey = (url: string, opts?: CachedBinaryFetchOptions) => { + if (!opts) + return hash(url) + // Vary on headers + redirect mode — callers with different user agents + // or redirect policies may get different upstream responses. + const parts = [url] + if (opts.headers) { + const entries = Object.entries(opts.headers).sort(([a], [b]) => a.localeCompare(b)) + for (const [k, v] of entries) + parts.push(`${k}=${v}`) + } + if (opts.redirect) + parts.push(`redirect=${opts.redirect}`) + if (opts.ignoreResponseError !== undefined) + parts.push(`ignoreResponseError=${opts.ignoreResponseError}`) + return hash(parts) + } const cached = defineCachedFunction( async (url: string, opts?: CachedBinaryFetchOptions): Promise => { const response = await fetchBoundedUpstream(url, { @@ -340,27 +431,16 @@ export function createCachedBinaryFetch( maxAge, swr: true, staleMaxAge: maxAge, - getKey: (url: string, opts?: CachedBinaryFetchOptions) => { - if (!opts) - return hash(url) - // Vary on headers + redirect mode — callers with different user agents - // or redirect policies may get different upstream responses. - const parts = [url] - if (opts.headers) { - const entries = Object.entries(opts.headers).sort(([a], [b]) => a.localeCompare(b)) - for (const [k, v] of entries) - parts.push(`${k}=${v}`) - } - if (opts.redirect) - parts.push(`redirect=${opts.redirect}`) - if (opts.ignoreResponseError !== undefined) - parts.push(`ignoreResponseError=${opts.ignoreResponseError}`) - return hash(parts) - }, + getKey: cacheKey, }, ) return async (url, opts) => { - const result = await cached(url, opts) + const key = cacheKey(url, opts) + failureGate.replay(key) + const result = await cached(url, opts).catch((error) => { + failureGate.record(key, error) + throw error + }) return { ...result, body: result.base64 ? Buffer.from(result.base64, 'base64') : Buffer.alloc(0), @@ -381,7 +461,9 @@ export function createCachedJsonFetch( config: CachedJsonFetchConfig, ): (url: string, opts?: { headers?: Record, timeout?: number }) => Promise { const maxResponseBytes = resolveMaxResponseBytes(config.maxResponseBytes, DEFAULT_JSON_MAX_RESPONSE_BYTES) - return defineCachedFunction( + const failureGate = createFailureGate(maxAge) + const cacheKey = (url: string, opts?: { headers?: Record }) => hash(getKey(url, opts)) + const cached = defineCachedFunction( async (url: string, opts?: { headers?: Record, timeout?: number }) => { const response = await fetchBoundedUpstream(url, { allowUrl: config.allowUrl, @@ -415,7 +497,15 @@ export function createCachedJsonFetch( maxAge, swr: true, staleMaxAge: maxAge, - getKey: (url, opts) => hash(getKey(url, opts)), + getKey: cacheKey, }, ) + return async (url, opts) => { + const key = cacheKey(url, opts) + failureGate.replay(key) + return cached(url, opts).catch((error) => { + failureGate.record(key, error) + throw error + }) + } } diff --git a/test/unit/cached-upstream-failure.test.ts b/test/unit/cached-upstream-failure.test.ts new file mode 100644 index 00000000..699b1cca --- /dev/null +++ b/test/unit/cached-upstream-failure.test.ts @@ -0,0 +1,132 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { closeMock, rawFetchMock } = vi.hoisted(() => ({ + closeMock: vi.fn(), + rawFetchMock: vi.fn(), +})) + +// Nitro caches a resolved value and stores nothing when the resolver throws. +// This stand-in keeps that contract so the failure gate is what is under test. +vi.mock('#nuxt-scripts/nitro', () => ({ + defineCachedFunction: (handler: (...args: any[]) => any, options: any) => { + const store = new Map() + return async (...args: any[]) => { + const key = options.getKey(...args) + if (store.has(key)) + return store.get(key) + const value = await handler(...args) + store.set(key, value) + return value + } + }, + useRuntimeConfig: () => ({}), +})) + +vi.mock('ofetch', () => ({ + createFetch: vi.fn(() => Object.assign(vi.fn(), { raw: rawFetchMock })), +})) + +vi.mock('../../packages/script/src/runtime/server/utils/network-host', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + createPublicNetworkDispatcher: async () => ({ fetch: globalThis.fetch, close: closeMock }), + } +}) + +const { createCachedJsonFetch, UPSTREAM_FAILURE_MAX_AGE } = await import( + '../../packages/script/src/runtime/server/utils/cached-upstream', +) + +function responseStream(body: string): ReadableStream { + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)) + controller.close() + }, + }) +} + +function rateLimited() { + return { + _data: responseStream('rate limited'), + headers: new Headers({ 'content-type': 'application/json' }), + status: 503, + statusText: 'Service Unavailable', + } +} + +function profile() { + return { + _data: responseStream('{"did":"did:plc:example"}'), + headers: new Headers({ 'content-type': 'application/json' }), + status: 200, + } +} + +beforeEach(() => { + vi.useFakeTimers() + closeMock.mockReset().mockResolvedValue(undefined) + rawFetchMock.mockReset() +}) + +afterEach(() => { + vi.useRealTimers() +}) + +describe('upstream failure caching', () => { + it('serves a failing upstream from cache instead of re-fetching it', async () => { + rawFetchMock.mockResolvedValue(rateLimited()) + const fetchProfile = createCachedJsonFetch('profile', 600, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + contentTypePrefixes: ['application/json'], + }) + + for (let i = 0; i < 3; i++) { + await expect(fetchProfile('https://public.api.bsky.app/profile')) + .rejects + .toMatchObject({ statusCode: 502, statusMessage: 'Upstream request failed' }) + } + + expect(rawFetchMock).toHaveBeenCalledOnce() + }) + + it('retries the upstream once the failure window closes', async () => { + rawFetchMock.mockResolvedValueOnce(rateLimited()).mockResolvedValueOnce(profile()) + const fetchProfile = createCachedJsonFetch<{ did: string }>('profile', 600, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + contentTypePrefixes: ['application/json'], + }) + + await expect(fetchProfile('https://public.api.bsky.app/profile')).rejects.toMatchObject({ statusCode: 502 }) + vi.advanceTimersByTime(UPSTREAM_FAILURE_MAX_AGE * 1000 + 1) + + await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) + expect(rawFetchMock).toHaveBeenCalledTimes(2) + }) + + it('caches a failure for no longer than the success it replaces', async () => { + rawFetchMock.mockResolvedValueOnce(rateLimited()).mockResolvedValueOnce(profile()) + const fetchProfile = createCachedJsonFetch<{ did: string }>('profile', 10, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + contentTypePrefixes: ['application/json'], + }) + + await expect(fetchProfile('https://public.api.bsky.app/profile')).rejects.toMatchObject({ statusCode: 502 }) + vi.advanceTimersByTime(11 * 1000) + + await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) + }) + + it('keeps serving a cached success after the upstream starts failing', async () => { + rawFetchMock.mockResolvedValueOnce(profile()).mockResolvedValue(rateLimited()) + const fetchProfile = createCachedJsonFetch<{ did: string }>('profile', 600, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + contentTypePrefixes: ['application/json'], + }) + + await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) + await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) + expect(rawFetchMock).toHaveBeenCalledOnce() + }) +}) diff --git a/test/unit/cached-upstream.test.ts b/test/unit/cached-upstream.test.ts index 39ed1b9a..7959a9ca 100644 --- a/test/unit/cached-upstream.test.ts +++ b/test/unit/cached-upstream.test.ts @@ -126,10 +126,69 @@ describe('upstream response bounds', () => { allowUrl: url => url.hostname === 'cdn.example.com', }) - await expect(fetchBinary('https://cdn.example.com/image')).rejects.toBe(upstreamFailure) + await expect(fetchBinary('https://cdn.example.com/image')) + .rejects + .toMatchObject({ statusCode: 502, message: 'upstream failed' }) expect(upstreamFailure).toMatchObject({ cleanupError: cleanupFailure }) }) + it('reports an upstream 5xx as a gateway failure instead of its own', async () => { + rawFetchMock.mockResolvedValueOnce({ + _data: responseStream('upstream is down'), + headers: new Headers({ 'content-type': 'application/json' }), + status: 503, + statusText: 'Service Unavailable', + }) + const fetchJson = createCachedJsonFetch('profile', 60, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + }) + + await expect(fetchJson('https://public.api.bsky.app/profile')) + .rejects + .toMatchObject({ statusCode: 502, statusMessage: 'Upstream request failed' }) + }) + + it('keeps an upstream 4xx as the client error it is', async () => { + rawFetchMock.mockResolvedValueOnce({ + _data: responseStream('slow down'), + headers: new Headers({ 'content-type': 'application/json' }), + status: 429, + statusText: 'Too Many Requests', + }) + const fetchJson = createCachedJsonFetch('profile', 60, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + }) + + await expect(fetchJson('https://public.api.bsky.app/profile')) + .rejects + .toMatchObject({ statusCode: 429, statusMessage: 'Too Many Requests' }) + }) + + it('reports a transport failure as a gateway failure', async () => { + rawFetchMock.mockRejectedValueOnce(Object.assign(new Error('connect ECONNREFUSED'), { name: 'FetchError' })) + const fetchJson = createCachedJsonFetch('profile', 60, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + }) + + await expect(fetchJson('https://public.api.bsky.app/profile')) + .rejects + .toMatchObject({ statusCode: 502, statusMessage: 'Upstream request failed' }) + }) + + it('reports an aborted upstream request as a gateway timeout', async () => { + rawFetchMock.mockRejectedValueOnce(Object.assign(new Error('request aborted'), { + name: 'FetchError', + cause: Object.assign(new Error('timed out'), { name: 'TimeoutError' }), + })) + const fetchJson = createCachedJsonFetch('profile', 60, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + }) + + await expect(fetchJson('https://public.api.bsky.app/profile')) + .rejects + .toMatchObject({ statusCode: 504, statusMessage: 'Gateway Timeout' }) + }) + it('rejects cached JSON over its configured byte limit', async () => { rawFetchMock.mockResolvedValueOnce({ _data: responseStream('{"value":"too large"}'), From b5776cf14204859be053759d5619ff1231f53e0d Mon Sep 17 00:00:00 2001 From: Harlan Wilton Date: Tue, 18 Aug 2026 13:34:59 +1000 Subject: [PATCH 2/3] test(script): cover proxy upstream failures against real transport Adds three checks the mocked tests could not make: - real sockets, real undici, real ofetch against a local upstream, so the timeout and refused-connection mappings are proved against the error objects production sees rather than hand-built ones; - the embed handlers mounted in h3, so the status the client receives is asserted end to end; - a stale-while-revalidate model taken from Nitro's cache runtime, so a failing upstream cannot turn a working stale embed into a 502. Moves the failure gate inside the cached resolver. A replayed failure now leaves the cache entry untouched, so it can never mask a stale success. --- .../runtime/server/utils/cached-upstream.ts | 42 +++-- test/unit/cached-upstream-failure.test.ts | 57 +++++- test/unit/cached-upstream-transport.test.ts | 169 ++++++++++++++++++ .../embed-handler-upstream-status.test.ts | 141 +++++++++++++++ 4 files changed, 383 insertions(+), 26 deletions(-) create mode 100644 test/unit/cached-upstream-transport.test.ts create mode 100644 test/unit/embed-handler-upstream-status.test.ts diff --git a/packages/script/src/runtime/server/utils/cached-upstream.ts b/packages/script/src/runtime/server/utils/cached-upstream.ts index 997a10e2..85d8ef38 100644 --- a/packages/script/src/runtime/server/utils/cached-upstream.ts +++ b/packages/script/src/runtime/server/utils/cached-upstream.ts @@ -139,8 +139,11 @@ function isTimeoutError(error: unknown): boolean { * Nitro stores nothing when the resolver throws, so an upstream that keeps * refusing a resource (rate limit, login wall, deleted post) is re-fetched on * every request. Each attempt raises a server error, and the retries deepen the - * rate limit that caused them. Stale-while-revalidate already covers a resource - * that was fetched successfully once, so this only gates the cold path. + * rate limit that caused them. + * + * The gate sits inside the cached resolver, so a replayed failure leaves the + * cache untouched. A resource that was fetched successfully once is still + * served stale by stale-while-revalidate while its upstream is down. */ function createFailureGate(maxAge: number) { const failures = new Map() @@ -409,6 +412,8 @@ export function createCachedBinaryFetch( } const cached = defineCachedFunction( async (url: string, opts?: CachedBinaryFetchOptions): Promise => { + const key = cacheKey(url, opts) + failureGate.replay(key) const response = await fetchBoundedUpstream(url, { allowContentType: config.allowContentType, allowUrl: config.allowUrl, @@ -418,6 +423,9 @@ export function createCachedBinaryFetch( maxResponseBytes, redirect: opts?.redirect ?? (config.allowUrl ? 'follow' : 'manual'), timeoutMs: opts?.timeout ?? 10000, + }).catch((error) => { + failureGate.record(key, error) + throw error }) return { base64: response.data.byteLength ? Buffer.from(response.data).toString('base64') : '', @@ -435,12 +443,7 @@ export function createCachedBinaryFetch( }, ) return async (url, opts) => { - const key = cacheKey(url, opts) - failureGate.replay(key) - const result = await cached(url, opts).catch((error) => { - failureGate.record(key, error) - throw error - }) + const result = await cached(url, opts) return { ...result, body: result.base64 ? Buffer.from(result.base64, 'base64') : Buffer.alloc(0), @@ -463,8 +466,10 @@ export function createCachedJsonFetch( const maxResponseBytes = resolveMaxResponseBytes(config.maxResponseBytes, DEFAULT_JSON_MAX_RESPONSE_BYTES) const failureGate = createFailureGate(maxAge) const cacheKey = (url: string, opts?: { headers?: Record }) => hash(getKey(url, opts)) - const cached = defineCachedFunction( + return defineCachedFunction( async (url: string, opts?: { headers?: Record, timeout?: number }) => { + const key = cacheKey(url, opts) + failureGate.replay(key) const response = await fetchBoundedUpstream(url, { allowUrl: config.allowUrl, contentTypePrefixes: config.contentTypePrefixes, @@ -474,6 +479,9 @@ export function createCachedJsonFetch( maxResponseBytes, redirect: 'follow', timeoutMs: opts?.timeout ?? 10000, + }).catch((error) => { + failureGate.record(key, error) + throw error }) const text = new TextDecoder().decode(response.data) let data: T @@ -488,7 +496,13 @@ export function createCachedJsonFetch( throw upstreamError('Upstream response is not valid JSON', 502, 'Invalid upstream response', cause) } } - config.validateResponse?.(data) + try { + config.validateResponse?.(data) + } + catch (error) { + failureGate.record(key, error) + throw error + } return data }, { @@ -500,12 +514,4 @@ export function createCachedJsonFetch( getKey: cacheKey, }, ) - return async (url, opts) => { - const key = cacheKey(url, opts) - failureGate.replay(key) - return cached(url, opts).catch((error) => { - failureGate.record(key, error) - throw error - }) - } } diff --git a/test/unit/cached-upstream-failure.test.ts b/test/unit/cached-upstream-failure.test.ts index 699b1cca..23770421 100644 --- a/test/unit/cached-upstream-failure.test.ts +++ b/test/unit/cached-upstream-failure.test.ts @@ -5,18 +5,35 @@ const { closeMock, rawFetchMock } = vi.hoisted(() => ({ rawFetchMock: vi.fn(), })) -// Nitro caches a resolved value and stores nothing when the resolver throws. -// This stand-in keeps that contract so the failure gate is what is under test. +/** + * Stand-in for Nitro's `defineCachedFunction`, following the three parts of its + * contract this module depends on (nitropack `runtime/internal/cache.mjs`): + * + * 1. a resolved value is stored, and a thrown resolver stores nothing; + * 2. a fresh entry is served without calling the resolver; + * 3. under `swr` a stale entry is returned straight away while the resolver + * refreshes in the background, and a failed refresh is swallowed. + */ vi.mock('#nuxt-scripts/nitro', () => ({ defineCachedFunction: (handler: (...args: any[]) => any, options: any) => { - const store = new Map() + const store = new Map() return async (...args: any[]) => { const key = options.getKey(...args) - if (store.has(key)) - return store.get(key) - const value = await handler(...args) - store.set(key, value) - return value + const entry = store.get(key) + const expired = !entry || Date.now() - entry.mtime > options.maxAge * 1000 + const resolve = expired + ? handler(...args).then((value: unknown) => { + store.set(key, { value, mtime: Date.now() }) + return value + }) + : Promise.resolve(entry!.value) + + if (entry && options.swr) { + // Nitro swallows a failed background refresh and keeps the stale entry. + resolve.catch((error: unknown) => error) + return entry.value + } + return resolve } }, useRuntimeConfig: () => ({}), @@ -129,4 +146,28 @@ describe('upstream failure caching', () => { await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) expect(rawFetchMock).toHaveBeenCalledOnce() }) + + it('still serves a stale success while the upstream is failing', async () => { + rawFetchMock.mockResolvedValueOnce(profile()).mockResolvedValue(rateLimited()) + const fetchProfile = createCachedJsonFetch<{ did: string }>('profile', 600, url => url, { + allowUrl: url => url.hostname === 'public.api.bsky.app', + contentTypePrefixes: ['application/json'], + }) + + await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) + + // Past the cache window, so every later call refreshes in the background + // and every refresh now fails. + for (let i = 0; i < 4; i++) { + vi.advanceTimersByTime(601 * 1000) + await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) + await Promise.resolve() + } + + // The failure gate must not turn a working stale embed into a 502. + vi.advanceTimersByTime(601 * 1000) + await expect(fetchProfile('https://public.api.bsky.app/profile')).resolves.toEqual({ did: 'did:plc:example' }) + // One success plus a failed background refresh for every stale read. + expect(rawFetchMock).toHaveBeenCalledTimes(6) + }) }) diff --git a/test/unit/cached-upstream-transport.test.ts b/test/unit/cached-upstream-transport.test.ts new file mode 100644 index 00000000..98fdb377 --- /dev/null +++ b/test/unit/cached-upstream-transport.test.ts @@ -0,0 +1,169 @@ +import type { Server } from 'node:http' +import type { AddressInfo } from 'node:net' +import { createServer } from 'node:http' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { closeMock } = vi.hoisted(() => ({ closeMock: vi.fn() })) + +// Nitro caches a resolved value and stores nothing when the resolver throws. +vi.mock('#nuxt-scripts/nitro', () => ({ + defineCachedFunction: (handler: (...args: any[]) => any, options: any) => { + const store = new Map() + return async (...args: any[]) => { + const key = options.getKey(...args) + if (store.has(key)) + return store.get(key) + const value = await handler(...args) + store.set(key, value) + return value + } + }, + useRuntimeConfig: () => ({}), +})) + +// Only the private-address guard is replaced: it refuses loopback, which is +// where the test upstream lives. ofetch, undici, and the sockets are real, so +// the errors under test are the ones production sees. +vi.mock('../../packages/script/src/runtime/server/utils/network-host', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + createPublicNetworkDispatcher: async () => ({ fetch: globalThis.fetch, close: closeMock }), + } +}) + +const { createCachedJsonFetch } = await import( + '../../packages/script/src/runtime/server/utils/cached-upstream', +) + +let server: Server +let origin: string +let requests: string[] + +function startServer(handle: (url: string, res: Parameters[0]>[1]) => void): Promise { + return new Promise((resolve) => { + server = createServer((req, res) => { + requests.push(req.url || '') + handle(req.url || '', res) + }) + server.listen(0, '127.0.0.1', () => { + origin = `http://127.0.0.1:${(server.address() as AddressInfo).port}` + resolve() + }) + }) +} + +function jsonFetch(maxAge = 600) { + return createCachedJsonFetch<{ ok: boolean }>('transport', maxAge, url => url, { + allowUrl: url => url.hostname === '127.0.0.1', + contentTypePrefixes: ['application/json'], + }) +} + +beforeEach(() => { + requests = [] + closeMock.mockReset().mockResolvedValue(undefined) +}) + +afterEach(async () => { + await new Promise(resolve => server?.close(() => resolve())) +}) + +describe('real upstream transport failures', () => { + it('reports a real upstream 503 as a gateway failure', async () => { + await startServer((_url, res) => { + res.writeHead(503, { 'content-type': 'application/json' }) + res.end('{"error":"rate limited"}') + }) + + await expect(jsonFetch()(`${origin}/profile`)) + .rejects + .toMatchObject({ statusCode: 502, statusMessage: 'Upstream request failed' }) + }) + + it('reports a refused connection as a gateway failure', async () => { + await startServer((_url, res) => res.end('{}')) + const port = (server.address() as AddressInfo).port + await new Promise(resolve => server.close(() => resolve())) + + await expect(jsonFetch()(`http://127.0.0.1:${port}/profile`)) + .rejects + .toMatchObject({ statusCode: 502, statusMessage: 'Upstream request failed' }) + }) + + it('reports an upstream that never answers as a gateway timeout', async () => { + await startServer(() => { + // Never respond; the request must be cut off by its own timeout. + }) + + await expect(jsonFetch()(`${origin}/profile`, { timeout: 150 })) + .rejects + .toMatchObject({ statusCode: 504, statusMessage: 'Gateway Timeout' }) + }) + + it('reports an upstream that stalls mid-body as a gateway timeout', async () => { + await startServer((_url, res) => { + res.writeHead(200, { 'content-type': 'application/json' }) + res.write('{"ok"') + // Headers and a partial body, then silence. + }) + + await expect(jsonFetch()(`${origin}/profile`, { timeout: 150 })) + .rejects + .toMatchObject({ statusCode: 504 }) + }) + + it('stops re-fetching an upstream that keeps failing', async () => { + await startServer((_url, res) => { + res.writeHead(503, { 'content-type': 'application/json' }) + res.end('{"error":"rate limited"}') + }) + const fetchProfile = jsonFetch() + + for (let i = 0; i < 5; i++) + await expect(fetchProfile(`${origin}/profile`)).rejects.toMatchObject({ statusCode: 502 }) + + expect(requests).toHaveLength(1) + }) + + it('tries the upstream again once the failure window closes', async () => { + let failing = true + await startServer((_url, res) => { + if (failing) { + res.writeHead(503, { 'content-type': 'application/json' }) + res.end('{"error":"rate limited"}') + return + } + res.writeHead(200, { 'content-type': 'application/json' }) + res.end('{"ok":true}') + }) + // The replay window is capped by the cache's own maxAge, so a 1s cache + // gives a 1s window. + const fetchProfile = jsonFetch(1) + + await expect(fetchProfile(`${origin}/profile`)).rejects.toMatchObject({ statusCode: 502 }) + failing = false + await expect(fetchProfile(`${origin}/profile`)).rejects.toMatchObject({ statusCode: 502 }) + + await new Promise(resolve => setTimeout(resolve, 1100)) + + await expect(fetchProfile(`${origin}/profile`)).resolves.toEqual({ ok: true }) + expect(requests).toHaveLength(2) + }) + + it('does not gate a resource the upstream still serves', async () => { + await startServer((url, res) => { + if (url.startsWith('/missing')) { + res.writeHead(404, { 'content-type': 'application/json' }) + res.end('{"error":"gone"}') + return + } + res.writeHead(200, { 'content-type': 'application/json' }) + res.end('{"ok":true}') + }) + const fetchProfile = jsonFetch() + + await expect(fetchProfile(`${origin}/missing`)).rejects.toMatchObject({ statusCode: 404 }) + await expect(fetchProfile(`${origin}/profile`)).resolves.toEqual({ ok: true }) + }) +}) diff --git a/test/unit/embed-handler-upstream-status.test.ts b/test/unit/embed-handler-upstream-status.test.ts new file mode 100644 index 00000000..4fec8aec --- /dev/null +++ b/test/unit/embed-handler-upstream-status.test.ts @@ -0,0 +1,141 @@ +import type { Server } from 'node:http' +import { createServer } from 'node:http' +import { createApp, toNodeListener } from 'h3' +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' + +const { rawFetchMock } = vi.hoisted(() => ({ rawFetchMock: vi.fn() })) + +// Nitro caches a resolved value and stores nothing when the resolver throws. +vi.mock('#nuxt-scripts/nitro', () => ({ + defineCachedFunction: (handler: (...args: any[]) => any, options: any) => { + const store = new Map() + return async (...args: any[]) => { + const key = options.getKey(...args) + if (store.has(key)) + return store.get(key) + const value = await handler(...args) + store.set(key, value) + return value + } + }, + useRuntimeConfig: () => ({}), +})) + +vi.mock('ofetch', () => ({ + createFetch: vi.fn(() => Object.assign(vi.fn(), { raw: rawFetchMock })), +})) + +const blueskyHandler = (await import('../../packages/script/src/runtime/server/bluesky-embed')).default +const instagramHandler = (await import('../../packages/script/src/runtime/server/instagram-embed')).default + +function stream(body: string): ReadableStream { + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)) + controller.close() + }, + }) +} + +function upstreamStatus(status: number, statusText: string, contentType: string) { + return { + _data: stream('{"error":"upstream"}'), + headers: new Headers({ 'content-type': contentType }), + status, + statusText, + } +} + +// Instagram answers a request it will not render with a JS-only shell. +const EMBED_SHELL = '
' + +function serve(handler: any) { + const app = createApp() + app.use(handler) + return createServer(toNodeListener(app)) +} + +describe('embed handlers report upstream faults as gateway errors', () => { + let bluesky: Server + let instagram: Server + let blueskyPort: number + let instagramPort: number + + beforeAll(async () => { + bluesky = serve(blueskyHandler) + instagram = serve(instagramHandler) + await new Promise(resolve => bluesky.listen(0, '127.0.0.1', resolve)) + await new Promise(resolve => instagram.listen(0, '127.0.0.1', resolve)) + blueskyPort = (bluesky.address() as { port: number }).port + instagramPort = (instagram.address() as { port: number }).port + }) + + beforeEach(() => { + rawFetchMock.mockReset() + }) + + afterAll(async () => { + await new Promise(resolve => bluesky.close(() => resolve())) + await new Promise(resolve => instagram.close(() => resolve())) + }) + + function requestBluesky(handle = 'nuxt.com') { + const post = encodeURIComponent(`https://bsky.app/profile/${handle}/post/abc123`) + return fetch(`http://127.0.0.1:${blueskyPort}/_scripts/embed/bluesky?url=${post}`) + } + + function requestInstagram(slug = 'example') { + const post = encodeURIComponent(`https://www.instagram.com/p/${slug}/`) + return fetch(`http://127.0.0.1:${instagramPort}/_scripts/embed/instagram?url=${post}`) + } + + it('answers 502 when Bluesky is down rather than mirroring its 503', async () => { + rawFetchMock.mockResolvedValue(upstreamStatus(503, 'Service Unavailable', 'application/json')) + + expect((await requestBluesky()).status).toBe(502) + }) + + it('answers 502 when Bluesky cannot be reached at all', async () => { + rawFetchMock.mockRejectedValue(Object.assign(new Error('connect ECONNREFUSED'), { name: 'FetchError' })) + + expect((await requestBluesky('unreachable.test')).status).toBe(502) + }) + + it('answers 504 when Bluesky does not respond in time', async () => { + rawFetchMock.mockRejectedValue(Object.assign(new Error('aborted'), { + name: 'FetchError', + cause: Object.assign(new Error('timed out'), { name: 'TimeoutError' }), + })) + + expect((await requestBluesky('slow.test')).status).toBe(504) + }) + + it('answers 502 when Instagram serves its empty embed shell', async () => { + rawFetchMock.mockResolvedValue({ + _data: stream(EMBED_SHELL), + headers: new Headers({ 'content-type': 'text/html' }), + status: 200, + }) + + expect((await requestInstagram()).status).toBe(502) + }) + + it('stops asking Instagram for a post it keeps refusing', async () => { + rawFetchMock.mockResolvedValue({ + _data: stream(EMBED_SHELL), + headers: new Headers({ 'content-type': 'text/html' }), + status: 200, + }) + + for (let i = 0; i < 4; i++) + expect((await requestInstagram('refused')).status).toBe(502) + + expect(rawFetchMock).toHaveBeenCalledOnce() + }) + + it('keeps a Bluesky 429 as the rate limit it is', async () => { + rawFetchMock.mockResolvedValue(upstreamStatus(429, 'Too Many Requests', 'application/json')) + + expect((await requestBluesky('limited.test')).status).toBe(429) + }) +}) From b66e649e87844f37938bd8ddde36d8cf7d721cbf Mon Sep 17 00:00:00 2001 From: Harlan Wilton Date: Tue, 18 Aug 2026 13:52:21 +1000 Subject: [PATCH 3/3] fix(script): key the failure gate on the request timeout A timeout is a failure the gate replays. Without the timeout in its key, a caller that allows the upstream longer inherited a shorter caller's 504 for the rest of the window. The cache key is unchanged: a stored response is just as valid however long the caller was willing to wait for it. --- .../runtime/server/utils/cached-upstream.ts | 18 ++++++++++++++---- test/unit/cached-upstream-transport.test.ts | 16 ++++++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/packages/script/src/runtime/server/utils/cached-upstream.ts b/packages/script/src/runtime/server/utils/cached-upstream.ts index 85d8ef38..a602a351 100644 --- a/packages/script/src/runtime/server/utils/cached-upstream.ts +++ b/packages/script/src/runtime/server/utils/cached-upstream.ts @@ -89,6 +89,7 @@ interface BoundedUpstreamResponse { const DEFAULT_BINARY_MAX_RESPONSE_BYTES = 10 * 1024 * 1024 const DEFAULT_JSON_MAX_RESPONSE_BYTES = 2 * 1024 * 1024 const TIMEOUT_ERROR_NAMES = new Set(['AbortError', 'BodyTimeoutError', 'HeadersTimeoutError', 'TimeoutError']) +const DEFAULT_UPSTREAM_TIMEOUT_MS = 10000 const MAX_TRACKED_FAILURES = 512 /** How long a failed upstream fetch is replayed before the upstream is tried again. */ @@ -410,9 +411,15 @@ export function createCachedBinaryFetch( parts.push(`ignoreResponseError=${opts.ignoreResponseError}`) return hash(parts) } + // The gate replays a failure, and a timeout is one. A caller that allows the + // upstream longer must not inherit a shorter caller's 504, so the gate keys + // on the timeout as well. The cache key stays as it is: a stored response is + // just as valid however long the caller was willing to wait for it. + const gateKey = (url: string, opts?: CachedBinaryFetchOptions) => + `${cacheKey(url, opts)}:${opts?.timeout ?? DEFAULT_UPSTREAM_TIMEOUT_MS}` const cached = defineCachedFunction( async (url: string, opts?: CachedBinaryFetchOptions): Promise => { - const key = cacheKey(url, opts) + const key = gateKey(url, opts) failureGate.replay(key) const response = await fetchBoundedUpstream(url, { allowContentType: config.allowContentType, @@ -422,7 +429,7 @@ export function createCachedBinaryFetch( maxRedirects: config.maxRedirects, maxResponseBytes, redirect: opts?.redirect ?? (config.allowUrl ? 'follow' : 'manual'), - timeoutMs: opts?.timeout ?? 10000, + timeoutMs: opts?.timeout ?? DEFAULT_UPSTREAM_TIMEOUT_MS, }).catch((error) => { failureGate.record(key, error) throw error @@ -466,9 +473,12 @@ export function createCachedJsonFetch( const maxResponseBytes = resolveMaxResponseBytes(config.maxResponseBytes, DEFAULT_JSON_MAX_RESPONSE_BYTES) const failureGate = createFailureGate(maxAge) const cacheKey = (url: string, opts?: { headers?: Record }) => hash(getKey(url, opts)) + // See `createCachedBinaryFetch`: the gate keys on the timeout, the cache does not. + const gateKey = (url: string, opts?: { headers?: Record, timeout?: number }) => + `${cacheKey(url, opts)}:${opts?.timeout ?? DEFAULT_UPSTREAM_TIMEOUT_MS}` return defineCachedFunction( async (url: string, opts?: { headers?: Record, timeout?: number }) => { - const key = cacheKey(url, opts) + const key = gateKey(url, opts) failureGate.replay(key) const response = await fetchBoundedUpstream(url, { allowUrl: config.allowUrl, @@ -478,7 +488,7 @@ export function createCachedJsonFetch( maxRedirects: config.maxRedirects, maxResponseBytes, redirect: 'follow', - timeoutMs: opts?.timeout ?? 10000, + timeoutMs: opts?.timeout ?? DEFAULT_UPSTREAM_TIMEOUT_MS, }).catch((error) => { failureGate.record(key, error) throw error diff --git a/test/unit/cached-upstream-transport.test.ts b/test/unit/cached-upstream-transport.test.ts index 98fdb377..424c70f4 100644 --- a/test/unit/cached-upstream-transport.test.ts +++ b/test/unit/cached-upstream-transport.test.ts @@ -151,6 +151,22 @@ describe('real upstream transport failures', () => { expect(requests).toHaveLength(2) }) + it('does not replay a short caller\'s timeout to one that waits longer', async () => { + await startServer((_url, res) => { + setTimeout(() => { + res.writeHead(200, { 'content-type': 'application/json' }) + res.end('{"ok":true}') + }, 300) + }) + const fetchProfile = jsonFetch() + + await expect(fetchProfile(`${origin}/profile`, { timeout: 100 })) + .rejects + .toMatchObject({ statusCode: 504 }) + + await expect(fetchProfile(`${origin}/profile`, { timeout: 2000 })).resolves.toEqual({ ok: true }) + }) + it('does not gate a resource the upstream still serves', async () => { await startServer((url, res) => { if (url.startsWith('/missing')) {