fix(proxy): report upstream failures as gateway errors (#862)

Author: harlan-zw

packages/script/src/runtime/server/utils/cached-upstream.ts

@@ -88,6 +88,12 @@ interface BoundedUpstreamResponse {
 
 constDEFAULT_BINARY_MAX_RESPONSE_BYTES=10*1024*1024
 constDEFAULT_JSON_MAX_RESPONSE_BYTES=2*1024*1024
+constTIMEOUT_ERROR_NAMES=newSet(['AbortError','BodyTimeoutError','HeadersTimeoutError','TimeoutError'])
+constDEFAULT_UPSTREAM_TIMEOUT_MS=10000
+constMAX_TRACKED_FAILURES=512
+
+/** How long a failed upstream fetch is replayed before the upstream is tried again. */
+exportconstUPSTREAM_FAILURE_MAX_AGE=60
 
 exportfunctionisSafeHttpsUrl(url: URL): boolean{
 returnurl.protocol==='https:'
@@ -104,6 +110,74 @@ 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.
+ */
+functionasUpstreamError(error: unknown): Error{
+if(typeof(erroras{statusCode?: unknown}|null)?.statusCode==='number')
+returnerrorasError
+consttimedOut=isTimeoutError(error)
+returnupstreamError(
+`Upstream request failed: ${(errorasError|null)?.message||'unknown error'}`,
+timedOut ? 504 : 502,
+timedOut ? 'Gateway Timeout' : 'Upstream request failed',
+error,
+)
+}
+
+functionisTimeoutError(error: unknown): boolean{
+constcandidate=erroras{name?: string,cause?: {name?: string,code?: string}}|null
+constcode=candidate?.cause?.code
+returnTIMEOUT_ERROR_NAMES.has(candidate?.name||'')
+||TIMEOUT_ERROR_NAMES.has(candidate?.cause?.name||'')
+||(typeofcode==='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.
+ *
+ * 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.
+ */
+functioncreateFailureGate(maxAge: number){
+constfailures=newMap<string,{until: number,error: {message: string,statusCode: number,statusMessage: string}}>()
+constfailureWindow=Math.min(UPSTREAM_FAILURE_MAX_AGE,maxAge)*1000
+
+return{
+replay(key: string): void{
+constfailure=failures.get(key)
+if(!failure)
+return
+if(Date.now()>=failure.until){
+failures.delete(key)
+return
+}
+throwupstreamError(failure.error.message,failure.error.statusCode,failure.error.statusMessage)
+},
+record(key: string,error: unknown): void{
+constfailure=errorasError&{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',
+},
+})
+},
+}
+}
+
 functionresolveMaxResponseBytes(value: number|undefined,fallback: number): number{
 constmaxBytes=value??fallback
 if(!Number.isSafeInteger(maxBytes)||maxBytes<0)
@@ -264,10 +338,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.
+constupstreamFault=response.status>=500
 awaitrejectResponse(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 +375,8 @@ async function fetchBoundedUpstream(
 }
 }
 catch(error){
-primaryError=error
-throwerror
+primaryError=asUpstreamError(error)
+throwprimaryError
 }
 finally{
 awaitclosePublicNetworkDispatcher(network,primaryError)
@@ -316,8 +393,34 @@ export function createCachedBinaryFetch(
 config: CachedBinaryFetchConfig={},
 ): (url: string,opts?: CachedBinaryFetchOptions)=>Promise<CachedBinaryResult>{
 constmaxResponseBytes=resolveMaxResponseBytes(config.maxResponseBytes,DEFAULT_BINARY_MAX_RESPONSE_BYTES)
+constfailureGate=createFailureGate(maxAge)
+constcacheKey=(url: string,opts?: CachedBinaryFetchOptions)=>{
+if(!opts)
+returnhash(url)
+// Vary on headers + redirect mode — callers with different user agents
+// or redirect policies may get different upstream responses.
+constparts=[url]
+if(opts.headers){
+constentries=Object.entries(opts.headers).sort(([a],[b])=>a.localeCompare(b))
+for(const[k,v]ofentries)
+parts.push(`${k}=${v}`)
+}
+if(opts.redirect)
+parts.push(`redirect=${opts.redirect}`)
+if(opts.ignoreResponseError!==undefined)
+parts.push(`ignoreResponseError=${opts.ignoreResponseError}`)
+returnhash(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.
+constgateKey=(url: string,opts?: CachedBinaryFetchOptions)=>
+`${cacheKey(url,opts)}:${opts?.timeout??DEFAULT_UPSTREAM_TIMEOUT_MS}`
 constcached=defineCachedFunction(
 async(url: string,opts?: CachedBinaryFetchOptions): Promise<CachedBinaryResponse&{status: number}>=>{
+constkey=gateKey(url,opts)
+failureGate.replay(key)
 constresponse=awaitfetchBoundedUpstream(url,{
 allowContentType: config.allowContentType,
 allowUrl: config.allowUrl,
@@ -326,7 +429,10 @@ 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)
+throwerror
 })
 return{
 base64: response.data.byteLength ? Buffer.from(response.data).toString('base64') : '',
@@ -340,23 +446,7 @@ export function createCachedBinaryFetch(
  maxAge,
 swr: true,
 staleMaxAge: maxAge,
-getKey: (url: string,opts?: CachedBinaryFetchOptions)=>{
-if(!opts)
-returnhash(url)
-// Vary on headers + redirect mode — callers with different user agents
-// or redirect policies may get different upstream responses.
-constparts=[url]
-if(opts.headers){
-constentries=Object.entries(opts.headers).sort(([a],[b])=>a.localeCompare(b))
-for(const[k,v]ofentries)
-parts.push(`${k}=${v}`)
-}
-if(opts.redirect)
-parts.push(`redirect=${opts.redirect}`)
-if(opts.ignoreResponseError!==undefined)
-parts.push(`ignoreResponseError=${opts.ignoreResponseError}`)
-returnhash(parts)
-},
+getKey: cacheKey,
 },
 )
 returnasync(url,opts)=>{
@@ -381,8 +471,15 @@ export function createCachedJsonFetch<T>(
 config: CachedJsonFetchConfig<T>,
 ): (url: string,opts?: {headers?: Record<string,string>,timeout?: number})=>Promise<T>{
 constmaxResponseBytes=resolveMaxResponseBytes(config.maxResponseBytes,DEFAULT_JSON_MAX_RESPONSE_BYTES)
+constfailureGate=createFailureGate(maxAge)
+constcacheKey=(url: string,opts?: {headers?: Record<string,string>})=>hash(getKey(url,opts))
+// See `createCachedBinaryFetch`: the gate keys on the timeout, the cache does not.
+constgateKey=(url: string,opts?: {headers?: Record<string,string>,timeout?: number})=>
+`${cacheKey(url,opts)}:${opts?.timeout??DEFAULT_UPSTREAM_TIMEOUT_MS}`
 returndefineCachedFunction(
 async(url: string,opts?: {headers?: Record<string,string>,timeout?: number})=>{
+constkey=gateKey(url,opts)
+failureGate.replay(key)
 constresponse=awaitfetchBoundedUpstream(url,{
 allowUrl: config.allowUrl,
 contentTypePrefixes: config.contentTypePrefixes,
@@ -391,7 +488,10 @@ export function createCachedJsonFetch<T>(
 maxRedirects: config.maxRedirects,
  maxResponseBytes,
 redirect: 'follow',
-timeoutMs: opts?.timeout??10000,
+timeoutMs: opts?.timeout??DEFAULT_UPSTREAM_TIMEOUT_MS,
+}).catch((error)=>{
+failureGate.record(key,error)
+throwerror
 })
 consttext=newTextDecoder().decode(response.data)
 letdata: T
@@ -406,7 +506,13 @@ export function createCachedJsonFetch<T>(
 throwupstreamError('Upstream response is not valid JSON',502,'Invalid upstream response',cause)
 }
 }
-config.validateResponse?.(data)
+try{
+config.validateResponse?.(data)
+}
+catch(error){
+failureGate.record(key,error)
+throwerror
+}
 returndata
 },
 {
@@ -415,7 +521,7 @@ export function createCachedJsonFetch<T>(
  maxAge,
 swr: true,
 staleMaxAge: maxAge,
-getKey: (url,opts)=>hash(getKey(url,opts)),
+getKey: cacheKey,
 },
 )
 }