diff --git a/packages/core/src/connection/http.ts b/packages/core/src/connection/http.ts index 2e4e5fa1..31d83b16 100644 --- a/packages/core/src/connection/http.ts +++ b/packages/core/src/connection/http.ts @@ -235,15 +235,29 @@ export interface HttpClient { externalGet: (externalUrl: string) => Promise; } -const fetchWithTimeout = ( +function fetchWithTimeout( input: string | URL | Request, timeout: number, init?: RequestInit | undefined -): Promise => { +): Promise; +function fetchWithTimeout( + input: string | URL | Request, + timeout: number, + init: RequestInit | undefined, + consumeResponse: (response: Response) => Promise +): Promise; +function fetchWithTimeout( + input: string | URL | Request, + timeout: number, + init?: RequestInit | undefined, + consumeResponse?: (response: Response) => Promise +): Promise { const controller = new AbortController(); // Set a timeout to abort the request const timeoutId = setTimeout(() => controller.abort(), timeout * 1000); + // Auth consumers include response bodies; other callers retain the header deadline. return fetch(input, { ...init, signal: controller.signal }) + .then((response) => (consumeResponse ? consumeResponse(response) : response)) .catch((error) => { if (isAbortError(error)) { throw new WeaviateRequestTimeoutError(`Request timed out after ${timeout}ms`); @@ -251,12 +265,13 @@ const fetchWithTimeout = ( throw error; // For other errors, rethrow them }) .finally(() => clearTimeout(timeoutId)); -}; +} export const httpClient = (config: InternalConnectionParams): HttpClient => { const version = '/v1'; const baseUri = `${config.host}${version}`; const url = makeUrl(baseUri); + const authTimeout = 5; return { close: () => config.agent?.destroy(), @@ -376,12 +391,17 @@ export const httpClient = (config: InternalConnectionParams): HttpClient => { return fetchWithTimeout(url(path), config.timeout?.query || 30, request); }, externalGet: (externalUrl: string) => { - return fetch(externalUrl, { - method: 'GET', - headers: { - ...config.headers, + return fetchWithTimeout( + externalUrl, + authTimeout, + { + method: 'GET', + headers: { + ...config.headers, + }, }, - }).then(checkStatus(true)); + checkStatus(true) + ); }, externalPost: (externalUrl: string, body: any, contentType: any) => { if (contentType == undefined || contentType == '') { @@ -398,7 +418,7 @@ export const httpClient = (config: InternalConnectionParams): HttpClient => { if (body != null) { request.body = body; } - return fetch(externalUrl, request).then(checkStatus(true)); + return fetchWithTimeout(externalUrl, authTimeout, request, checkStatus(true)); }, }; }; diff --git a/packages/test/node/connection/auth-timeout-body.test.ts b/packages/test/node/connection/auth-timeout-body.test.ts new file mode 100644 index 00000000..989594b3 --- /dev/null +++ b/packages/test/node/connection/auth-timeout-body.test.ts @@ -0,0 +1,138 @@ +import { createServer } from 'node:http'; +import { performance } from 'node:perf_hooks'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { httpClient } from '@weaviate/core/connection/http'; +import { + WeaviateInsufficientPermissionsError, + WeaviateRequestTimeoutError, + WeaviateUnauthenticatedError, + WeaviateUnexpectedStatusCodeError, +} from '@weaviate/core/errors'; + +const request = (kind: 'discovery' | 'token', url: string) => { + const client = httpClient({ host: 'http://weaviate.test' }); + return kind === 'discovery' + ? client.externalGet(url) + : client.externalPost( + url, + new URLSearchParams({ grant_type: 'client_credentials' }), + 'application/x-www-form-urlencoded' + ); +}; + +describe('OIDC response body deadlines', () => { + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + it.each(['discovery', 'token'] as const)( + 'keeps the %s deadline active until its body is consumed', + async (kind) => { + vi.useFakeTimers(); + let signal: AbortSignal | null | undefined; + vi.stubGlobal( + 'fetch', + vi.fn((_input: unknown, init?: RequestInit) => { + signal = init?.signal; + return Promise.resolve( + new Response( + new ReadableStream({ + start(controller) { + signal?.addEventListener('abort', () => controller.error(signal?.reason), { once: true }); + }, + }), + { headers: { 'content-type': 'application/json' } } + ) + ); + }) + ); + const result = request(kind, 'https://issuer.test/body').catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(4_999); + expect(signal?.aborted).toBe(false); + expect(vi.getTimerCount()).toBe(1); + await vi.advanceTimersByTimeAsync(1); + expect(signal?.aborted).toBe(true); + expect(await result).toBeInstanceOf(WeaviateRequestTimeoutError); + expect(vi.getTimerCount()).toBe(0); + } + ); + + it.each([ + [401, WeaviateUnauthenticatedError], + [403, WeaviateInsufficientPermissionsError], + [502, WeaviateUnexpectedStatusCodeError], + ] as const)('preserves HTTP %s errors and clears their deadline', async (status, errorType) => { + vi.useFakeTimers(); + vi.stubGlobal( + 'fetch', + vi.fn().mockImplementation(() => Promise.resolve(new Response('issuer refused', { status }))) + ); + await Promise.all( + (['discovery', 'token'] as const).map((kind) => + expect(request(kind, 'https://issuer.test/error')).rejects.toBeInstanceOf(errorType) + ) + ); + expect(vi.getTimerCount()).toBe(0); + }); +}); + +describe('real OIDC response body cancellation', () => { + it('aborts native discovery and token fetches after headers when JSON bodies stall', async () => { + const seen = new Set(); + const closed = new Set(); + let notifyClosed: () => void; + const allClosed = new Promise((resolve) => { + notifyClosed = resolve; + }); + const server = createServer((incoming, response) => { + const method = incoming.method!; + seen.add(method); + response.on('close', () => { + closed.add(method); + if (closed.size === 2) notifyClosed(); + }); + incoming.resume(); + response.writeHead(200, { 'content-type': 'application/json' }); + response.write('{"access_token":'); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + if (!address || typeof address === 'string') throw new Error('Missing loopback listener'); + const origin = `http://127.0.0.1:${address.port}`; + let guard: ReturnType | undefined; + let closeGuard: ReturnType | undefined; + try { + const started = performance.now(); + const results = await Promise.race([ + Promise.all( + (['discovery', 'token'] as const).map((kind) => + request(kind, origin).catch((error: unknown) => error) + ) + ), + new Promise((_resolve, reject) => { + guard = setTimeout( + () => reject(new Error('Auth response bodies remained pending past the bounded deadline')), + 8_000 + ); + }), + ]); + expect(seen).toEqual(new Set(['GET', 'POST'])); + for (const result of results) expect(result).toBeInstanceOf(WeaviateRequestTimeoutError); + expect(performance.now() - started).toBeGreaterThanOrEqual(4_500); + await Promise.race([ + allClosed, + new Promise((_resolve, reject) => { + closeGuard = setTimeout(() => reject(new Error('Timed-out auth bodies left sockets open')), 500); + }), + ]); + expect(closed).toEqual(new Set(['GET', 'POST'])); + } finally { + clearTimeout(guard); + clearTimeout(closeGuard); + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + } + }); +}); diff --git a/packages/test/node/connection/auth-timeout.test.ts b/packages/test/node/connection/auth-timeout.test.ts new file mode 100644 index 00000000..bb679760 --- /dev/null +++ b/packages/test/node/connection/auth-timeout.test.ts @@ -0,0 +1,57 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { WeaviateRequestTimeoutError } from '@weaviate/core/errors'; +import { httpClient } from '@weaviate/core/connection/http'; + +describe('OIDC HTTP request deadlines', () => { + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + const requests = { + discovery: () => httpClient({ host: 'http://weaviate.test' }).externalGet('https://issuer.test/openid'), + token: () => + httpClient({ host: 'http://weaviate.test' }).externalPost( + 'https://issuer.test/token', + new URLSearchParams({ grant_type: 'client_credentials' }), + 'application/x-www-form-urlencoded' + ), + }; + + it.each(Object.keys(requests) as (keyof typeof requests)[])( + 'aborts a stalled %s request after five seconds', + async (kind) => { + vi.useFakeTimers(); + let requestSignal: AbortSignal | null | undefined; + vi.stubGlobal( + 'fetch', + vi.fn((_input: unknown, init?: RequestInit) => { + requestSignal = init?.signal; + return new Promise((_resolve, reject) => { + requestSignal?.addEventListener('abort', () => reject(requestSignal?.reason), { once: true }); + }); + }) + ); + + const result = requests[kind]().catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(4_999); + expect(requestSignal?.aborted).not.toBe(true); + await vi.advanceTimersByTimeAsync(1); + expect(requestSignal?.aborted).toBe(true); + expect(await result).toBeInstanceOf(WeaviateRequestTimeoutError); + } + ); + + it.each(Object.keys(requests) as (keyof typeof requests)[])( + 'clears the deadline after a successful %s request', + async (kind) => { + vi.useFakeTimers(); + const payload = { access_token: 'example-token' }; + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(JSON.stringify(payload)))); + + expect(await requests[kind]()).toEqual(payload); + expect(vi.getTimerCount()).toBe(0); + } + ); +});