fix(api): retry transient session reads safely
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import test from 'node:test';
|
||||
|
||||
import { fetchWithTransientRetry } from './src/api/core.ts';
|
||||
|
||||
async function withMockFetch(mock, run) {
|
||||
const originalFetch = globalThis.fetch;
|
||||
globalThis.fetch = mock;
|
||||
try {
|
||||
await run();
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
}
|
||||
}
|
||||
|
||||
test('idempotent requests retry transient network failures', async () => {
|
||||
let calls = 0;
|
||||
await withMockFetch(
|
||||
async () => {
|
||||
calls += 1;
|
||||
if (calls < 3) throw new TypeError('fetch failed');
|
||||
return new Response('{}', { status: 200 });
|
||||
},
|
||||
async () => {
|
||||
const response = await fetchWithTransientRetry('/api/sessions', undefined, 1_000, [0, 0]);
|
||||
assert.equal(response.status, 200);
|
||||
},
|
||||
);
|
||||
assert.equal(calls, 3);
|
||||
});
|
||||
|
||||
test('idempotent requests retry transient gateway responses', async () => {
|
||||
let calls = 0;
|
||||
await withMockFetch(
|
||||
async () => {
|
||||
calls += 1;
|
||||
return calls === 1
|
||||
? new Response('temporary', { status: 503 })
|
||||
: new Response('{}', { status: 200 });
|
||||
},
|
||||
async () => {
|
||||
const response = await fetchWithTransientRetry('/api/sessions', undefined, 1_000, [0]);
|
||||
assert.equal(response.status, 200);
|
||||
},
|
||||
);
|
||||
assert.equal(calls, 2);
|
||||
});
|
||||
|
||||
test('non-idempotent requests are never retried', async () => {
|
||||
let calls = 0;
|
||||
await withMockFetch(
|
||||
async () => {
|
||||
calls += 1;
|
||||
throw new TypeError('fetch failed');
|
||||
},
|
||||
async () => {
|
||||
await assert.rejects(
|
||||
fetchWithTransientRetry(
|
||||
'/api/agent/start',
|
||||
{ method: 'POST', body: '{}' },
|
||||
1_000,
|
||||
[0, 0],
|
||||
),
|
||||
/fetch failed/,
|
||||
);
|
||||
},
|
||||
);
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
|
||||
test('aborted requests are never retried', async () => {
|
||||
let calls = 0;
|
||||
await withMockFetch(
|
||||
async () => {
|
||||
calls += 1;
|
||||
throw new DOMException('Aborted', 'AbortError');
|
||||
},
|
||||
async () => {
|
||||
await assert.rejects(
|
||||
fetchWithTransientRetry('/api/sessions', undefined, 1_000, [0, 0]),
|
||||
(error) => error instanceof DOMException && error.name === 'AbortError',
|
||||
);
|
||||
},
|
||||
);
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
+1
-1
File diff suppressed because one or more lines are too long
+9
-3
@@ -1443,7 +1443,9 @@ export async function listSessions(options?: {
|
||||
if (options?.offset != null) params.set('offset', String(options.offset));
|
||||
if (options?.query?.trim()) params.set('query', options.query.trim());
|
||||
const qs = params.size ? `?${params.toString()}` : '';
|
||||
const result = await apiFetch<SessionListResponse>(`/sessions${qs}`);
|
||||
const result = await apiFetch<SessionListResponse>(`/sessions${qs}`, undefined, {
|
||||
retryOnTransientFailure: true,
|
||||
});
|
||||
return { items: result.sessions ?? [], page: result.page ?? {} };
|
||||
}
|
||||
|
||||
@@ -1452,7 +1454,9 @@ function sessionPath(sessionId: string, suffix = '') {
|
||||
}
|
||||
|
||||
export async function getSession(sessionId: string): Promise<Session> {
|
||||
return apiFetch<Session>(sessionPath(sessionId));
|
||||
return apiFetch<Session>(sessionPath(sessionId), undefined, {
|
||||
retryOnTransientFailure: true,
|
||||
});
|
||||
}
|
||||
|
||||
export async function deleteChatSession(sessionId: string): Promise<void> {
|
||||
@@ -1474,7 +1478,9 @@ export async function loadSessionDetail(
|
||||
if (history?.before != null) params.set('history_before', String(history.before));
|
||||
if (history?.limit != null) params.set('history_limit', String(history.limit));
|
||||
const qs = params.size ? `?${params.toString()}` : '';
|
||||
const detail = await apiFetch<Session>(`${sessionPath(sessionId)}${qs}`);
|
||||
const detail = await apiFetch<Session>(`${sessionPath(sessionId)}${qs}`, undefined, {
|
||||
retryOnTransientFailure: true,
|
||||
});
|
||||
const messages = normalizeConversationMessages(
|
||||
(detail.conversation ?? []).filter((m) => m.metadata?.userVisible),
|
||||
);
|
||||
|
||||
+72
-13
@@ -2,6 +2,8 @@ import type { InsufficientBalanceDetails, SessionEvent } from '../types';
|
||||
|
||||
export const API = '/api';
|
||||
const DEFAULT_API_TIMEOUT_MS = 20_000;
|
||||
const DEFAULT_TRANSIENT_RETRY_DELAYS_MS = [500, 1_000] as const;
|
||||
const RETRYABLE_RESPONSE_STATUSES = new Set([502, 503, 504]);
|
||||
|
||||
export class ApiError extends Error {
|
||||
readonly status: number;
|
||||
@@ -123,7 +125,7 @@ export async function fetchWithTimeout(
|
||||
): Promise<Response> {
|
||||
const controller = new AbortController();
|
||||
const upstreamSignal = init?.signal;
|
||||
const timeout = window.setTimeout(() => controller.abort(), timeoutMs);
|
||||
const timeout = globalThis.setTimeout(() => controller.abort(), timeoutMs);
|
||||
|
||||
const abortFromUpstream = () => controller.abort();
|
||||
if (upstreamSignal) {
|
||||
@@ -137,11 +139,68 @@ export async function fetchWithTimeout(
|
||||
signal: controller.signal,
|
||||
});
|
||||
} finally {
|
||||
window.clearTimeout(timeout);
|
||||
globalThis.clearTimeout(timeout);
|
||||
upstreamSignal?.removeEventListener('abort', abortFromUpstream);
|
||||
}
|
||||
}
|
||||
|
||||
function isIdempotentRequest(init?: RequestInit) {
|
||||
const method = String(init?.method ?? 'GET').toUpperCase();
|
||||
return method === 'GET' || method === 'HEAD' || method === 'OPTIONS';
|
||||
}
|
||||
|
||||
function isRetryableNetworkError(err: unknown) {
|
||||
if (err instanceof DOMException && err.name === 'AbortError') return false;
|
||||
const message = err instanceof Error ? err.message : String(err ?? '');
|
||||
return /failed to fetch|networkerror|fetch failed|econn|enotfound|network/i.test(message);
|
||||
}
|
||||
|
||||
async function waitForTransientRetry(delayMs: number, signal?: AbortSignal) {
|
||||
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const finish = () => {
|
||||
signal?.removeEventListener('abort', abort);
|
||||
resolve();
|
||||
};
|
||||
const abort = () => {
|
||||
globalThis.clearTimeout(timeout);
|
||||
reject(new DOMException('Aborted', 'AbortError'));
|
||||
};
|
||||
const timeout = globalThis.setTimeout(finish, delayMs);
|
||||
signal?.addEventListener('abort', abort, { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
export async function fetchWithTransientRetry(
|
||||
input: RequestInfo | URL,
|
||||
init: RequestInit | undefined,
|
||||
timeoutMs: number,
|
||||
retryDelaysMs: readonly number[] = DEFAULT_TRANSIENT_RETRY_DELAYS_MS,
|
||||
): Promise<Response> {
|
||||
if (!isIdempotentRequest(init) || retryDelaysMs.length === 0) {
|
||||
return fetchWithTimeout(input, init, timeoutMs);
|
||||
}
|
||||
|
||||
let attempt = 0;
|
||||
while (true) {
|
||||
try {
|
||||
const response = await fetchWithTimeout(input, init, timeoutMs);
|
||||
if (
|
||||
!RETRYABLE_RESPONSE_STATUSES.has(response.status)
|
||||
|| attempt >= retryDelaysMs.length
|
||||
) {
|
||||
return response;
|
||||
}
|
||||
await response.body?.cancel().catch(() => undefined);
|
||||
} catch (err) {
|
||||
if (!isRetryableNetworkError(err) || attempt >= retryDelaysMs.length) throw err;
|
||||
}
|
||||
|
||||
await waitForTransientRetry(retryDelaysMs[attempt], init?.signal ?? undefined);
|
||||
attempt += 1;
|
||||
}
|
||||
}
|
||||
|
||||
export async function portalFetch<T>(path: string, init?: RequestInit): Promise<T> {
|
||||
let res: Response;
|
||||
try {
|
||||
@@ -206,21 +265,21 @@ export function sanitizeSessionEvent(event: SessionEvent): SessionEvent {
|
||||
export async function apiFetch<T>(
|
||||
path: string,
|
||||
init?: RequestInit,
|
||||
options?: { timeoutMs?: number },
|
||||
options?: { timeoutMs?: number; retryOnTransientFailure?: boolean },
|
||||
): Promise<T> {
|
||||
let res: Response;
|
||||
try {
|
||||
res = await fetchWithTimeout(
|
||||
`${API}${path}`,
|
||||
{
|
||||
...init,
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
...init?.headers,
|
||||
},
|
||||
const requestInit = {
|
||||
...init,
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
...init?.headers,
|
||||
},
|
||||
options?.timeoutMs,
|
||||
);
|
||||
};
|
||||
const timeoutMs = options?.timeoutMs ?? DEFAULT_API_TIMEOUT_MS;
|
||||
res = options?.retryOnTransientFailure
|
||||
? await fetchWithTransientRetry(`${API}${path}`, requestInit, timeoutMs)
|
||||
: await fetchWithTimeout(`${API}${path}`, requestInit, timeoutMs);
|
||||
} catch (err) {
|
||||
throw new ApiError(0, formatNetworkError(err));
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user