fix(web): kill RSC prefetch storm + dedupe in-flight fetches + retry on transient timeout (closes #1855) (#1856)
* fix: reduce session page fetch starvation * fix: address fetch dedupe review feedback * fix: preserve body-read timeouts in client fetch --------- Co-authored-by: i-trytoohard <193449657+i-trytoohard@users.noreply.github.com>
This commit is contained in:
i-trytoohard
GitHub
parent
3fb23cfb48
commit
e6ad078d7a
|
|
@ -149,20 +149,20 @@ describe("SessionPage project polling", () => {
|
|||
|
||||
expect(fetch).toHaveBeenCalledWith(
|
||||
"/api/projects",
|
||||
expect.objectContaining({ signal: expect.any(AbortSignal) }),
|
||||
expect.any(Object),
|
||||
);
|
||||
expect(fetch).toHaveBeenCalledWith(
|
||||
"/api/sessions/worker-1",
|
||||
expect.objectContaining({ signal: expect.any(AbortSignal) }),
|
||||
expect.any(Object),
|
||||
);
|
||||
expect(fetch).toHaveBeenCalledWith(
|
||||
"/api/sessions?fresh=true",
|
||||
expect.objectContaining({ signal: expect.any(AbortSignal) }),
|
||||
expect.any(Object),
|
||||
);
|
||||
|
||||
expect(fetch).toHaveBeenCalledWith(
|
||||
"/api/sessions?project=my-app&orchestratorOnly=true&fresh=true",
|
||||
expect.objectContaining({ signal: expect.any(AbortSignal) }),
|
||||
expect.any(Object),
|
||||
);
|
||||
|
||||
expect(
|
||||
|
|
@ -258,7 +258,7 @@ describe("SessionPage project polling", () => {
|
|||
|
||||
expect(fetch).toHaveBeenCalledWith(
|
||||
"/api/sessions?project=my-app&fresh=true",
|
||||
expect.objectContaining({ signal: expect.any(AbortSignal) }),
|
||||
expect.any(Object),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
@ -336,7 +336,7 @@ describe("SessionPage project polling", () => {
|
|||
expect(screen.getByRole("button", { name: "Toggle sidebar" })).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("times out a stuck session fetch and replaces the infinite loader with an error state", async () => {
|
||||
it("keeps retrying after a stuck session fetch times out before showing an error state", async () => {
|
||||
global.fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => {
|
||||
const url = String(input);
|
||||
if (url === "/api/projects") {
|
||||
|
|
@ -377,12 +377,30 @@ describe("SessionPage project polling", () => {
|
|||
});
|
||||
await flushAsyncWork();
|
||||
|
||||
expect(screen.getByText("Loading session…")).toBeInTheDocument();
|
||||
expect(screen.queryByText("Failed to load session")).not.toBeInTheDocument();
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
});
|
||||
await flushAsyncWork();
|
||||
|
||||
expect(screen.getByText("Loading session…")).toBeInTheDocument();
|
||||
expect(screen.queryByText("Failed to load session")).not.toBeInTheDocument();
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(12_000);
|
||||
});
|
||||
await flushAsyncWork();
|
||||
|
||||
expect(screen.getByText("Failed to load session")).toBeInTheDocument();
|
||||
expect(screen.getByText(/taking too long/i)).toBeInTheDocument();
|
||||
expect(screen.queryByText("Loading session…")).not.toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("shows a recoverable unavailable state when the first session request aborts", async () => {
|
||||
it("does not show the route error after the first aborted session request", async () => {
|
||||
let sessionFetches = 0;
|
||||
|
||||
global.fetch = vi.fn((input: RequestInfo | URL) => {
|
||||
const url = String(input);
|
||||
if (url === "/api/projects") {
|
||||
|
|
@ -396,6 +414,7 @@ describe("SessionPage project polling", () => {
|
|||
}
|
||||
|
||||
if (url === "/api/sessions/worker-1") {
|
||||
sessionFetches += 1;
|
||||
return Promise.reject(new DOMException("Aborted", "AbortError"));
|
||||
}
|
||||
|
||||
|
|
@ -415,9 +434,24 @@ describe("SessionPage project polling", () => {
|
|||
render(<SessionPage />);
|
||||
await flushAsyncWork();
|
||||
|
||||
expect(screen.getByText("Session unavailable")).toBeInTheDocument();
|
||||
expect(screen.getByText(/backend has not returned this session yet/i)).toBeInTheDocument();
|
||||
expect(screen.queryByText("Loading session…")).not.toBeInTheDocument();
|
||||
expect(sessionFetches).toBe(1);
|
||||
expect(screen.getByText("Loading session…")).toBeInTheDocument();
|
||||
expect(screen.queryByText("Failed to load session")).not.toBeInTheDocument();
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(3_000);
|
||||
});
|
||||
await flushAsyncWork();
|
||||
|
||||
expect(sessionFetches).toBeGreaterThanOrEqual(3);
|
||||
expect(screen.queryByText("Failed to load session")).not.toBeInTheDocument();
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(4_000);
|
||||
});
|
||||
await flushAsyncWork();
|
||||
|
||||
expect(screen.getAllByText("Failed to load session").length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("marks sidebar data as loading until the sessions list resolves", async () => {
|
||||
|
|
|
|||
|
|
@ -84,6 +84,9 @@ let cachedProjects: ProjectInfo[] | null = null;
|
|||
let cachedSidebarSessions: DashboardSession[] | null = null;
|
||||
const SESSION_PAGE_REFRESH_INTERVAL_MS = 2000;
|
||||
const SESSION_FETCH_TIMEOUT_MS = 8000;
|
||||
const SESSION_LOAD_MAX_CONSECUTIVE_FAILURES = 4;
|
||||
const SESSION_LOAD_MAX_RETRY_ELAPSED_MS = 30_000;
|
||||
const SESSION_LOAD_RETRY_BACKOFF_MS = [1_000, 2_000, 4_000] as const;
|
||||
const PROJECT_SIDEBAR_FETCH_TIMEOUT_MS = 5000;
|
||||
const PROJECTS_FETCH_TIMEOUT_MS = 5000;
|
||||
const validSessionStatuses = new Set<string>(Object.values(SESSION_STATUS));
|
||||
|
|
@ -136,6 +139,19 @@ function isAbortLikeError(error: unknown): boolean {
|
|||
return false;
|
||||
}
|
||||
|
||||
function isTransientSessionLoadError(error: unknown): boolean {
|
||||
if (isAbortLikeError(error)) return true;
|
||||
if (error instanceof Error) {
|
||||
const message = error.message.toLowerCase();
|
||||
return (
|
||||
message.includes("timed out") ||
|
||||
message.includes("network") ||
|
||||
message.includes("failed to fetch")
|
||||
);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function getSessionLoadErrorMessage(error: Error): string {
|
||||
const normalized = error.message.toLowerCase();
|
||||
if (normalized.includes("timed out")) {
|
||||
|
|
@ -418,6 +434,22 @@ export default function SessionPage() {
|
|||
const projectSessionsFetchControllerRef = useRef<AbortController | null>(null);
|
||||
const sidebarFetchControllerRef = useRef<AbortController | null>(null);
|
||||
const pageUnloadingRef = useRef(false);
|
||||
const sessionLoadFailureCountRef = useRef(0);
|
||||
const sessionLoadFirstFailureAtRef = useRef<number | null>(null);
|
||||
const sessionLoadRetryTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
||||
|
||||
const clearSessionLoadRetry = useCallback(() => {
|
||||
if (sessionLoadRetryTimerRef.current) {
|
||||
clearTimeout(sessionLoadRetryTimerRef.current);
|
||||
sessionLoadRetryTimerRef.current = null;
|
||||
}
|
||||
}, []);
|
||||
|
||||
const resetSessionLoadFailures = useCallback(() => {
|
||||
sessionLoadFailureCountRef.current = 0;
|
||||
sessionLoadFirstFailureAtRef.current = null;
|
||||
clearSessionLoadRetry();
|
||||
}, [clearSessionLoadRetry]);
|
||||
|
||||
// Keep prefixByProjectRef in sync so fetchProjectSessions (stable [] dep) reads latest map
|
||||
useEffect(() => {
|
||||
|
|
@ -497,6 +529,7 @@ export default function SessionPage() {
|
|||
fetchingSessionRef.current = true;
|
||||
const controller = new AbortController();
|
||||
sessionFetchControllerRef.current = controller;
|
||||
let keepLoadingForRetry = false;
|
||||
try {
|
||||
const data = await fetchJsonWithTimeout<DashboardSession | { error: string }>(
|
||||
`/api/sessions/${encodeURIComponent(id)}`,
|
||||
|
|
@ -510,8 +543,9 @@ export default function SessionPage() {
|
|||
setRouteError(null);
|
||||
setSessionMissing(false);
|
||||
hasLoadedSessionRef.current = true;
|
||||
resetSessionLoadFailures();
|
||||
} catch (err) {
|
||||
if (pageUnloadingRef.current || controller.signal.aborted || isAbortLikeError(err)) {
|
||||
if (pageUnloadingRef.current || controller.signal.aborted) {
|
||||
return;
|
||||
}
|
||||
const message = err instanceof Error ? err.message : "Failed to load session";
|
||||
|
|
@ -524,20 +558,55 @@ export default function SessionPage() {
|
|||
setSessionMissing(true);
|
||||
}
|
||||
setLoading(false);
|
||||
resetSessionLoadFailures();
|
||||
return;
|
||||
}
|
||||
|
||||
if (!hasLoadedSessionRef.current && isTransientSessionLoadError(err)) {
|
||||
const failureCount = sessionLoadFailureCountRef.current + 1;
|
||||
sessionLoadFailureCountRef.current = failureCount;
|
||||
sessionLoadFirstFailureAtRef.current ??= Date.now();
|
||||
const elapsedMs = Date.now() - sessionLoadFirstFailureAtRef.current;
|
||||
const shouldKeepRetrying =
|
||||
failureCount < SESSION_LOAD_MAX_CONSECUTIVE_FAILURES &&
|
||||
elapsedMs < SESSION_LOAD_MAX_RETRY_ELAPSED_MS;
|
||||
|
||||
if (shouldKeepRetrying) {
|
||||
const delay =
|
||||
SESSION_LOAD_RETRY_BACKOFF_MS[
|
||||
Math.min(failureCount - 1, SESSION_LOAD_RETRY_BACKOFF_MS.length - 1)
|
||||
];
|
||||
keepLoadingForRetry = true;
|
||||
setLoading(true);
|
||||
console.warn("Session fetch failed transiently; retrying", {
|
||||
sessionId: id,
|
||||
failureCount,
|
||||
retryInMs: delay,
|
||||
error: err,
|
||||
});
|
||||
clearSessionLoadRetry();
|
||||
sessionLoadRetryTimerRef.current = setTimeout(() => {
|
||||
sessionLoadRetryTimerRef.current = null;
|
||||
void fetchSession();
|
||||
}, delay);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
console.error("Failed to fetch session:", err);
|
||||
if (!hasLoadedSessionRef.current) {
|
||||
setRouteError(err instanceof Error ? err : new Error("Failed to load session"));
|
||||
}
|
||||
} finally {
|
||||
setLoading(false);
|
||||
if (!keepLoadingForRetry) {
|
||||
setLoading(false);
|
||||
}
|
||||
fetchingSessionRef.current = false;
|
||||
if (sessionFetchControllerRef.current === controller) {
|
||||
sessionFetchControllerRef.current = null;
|
||||
}
|
||||
}
|
||||
}, [id]);
|
||||
}, [clearSessionLoadRetry, id, resetSessionLoadFailures]);
|
||||
|
||||
const fetchProjectSessions = useCallback(async () => {
|
||||
if (fetchingProjectSessionsRef.current) return;
|
||||
|
|
@ -727,11 +796,12 @@ export default function SessionPage() {
|
|||
|
||||
useEffect(() => {
|
||||
return () => {
|
||||
clearSessionLoadRetry();
|
||||
sessionFetchControllerRef.current?.abort();
|
||||
projectSessionsFetchControllerRef.current?.abort();
|
||||
sidebarFetchControllerRef.current?.abort();
|
||||
};
|
||||
}, []);
|
||||
}, [clearSessionLoadRetry]);
|
||||
|
||||
if (loading) {
|
||||
return (
|
||||
|
|
@ -816,6 +886,7 @@ export default function SessionPage() {
|
|||
setRouteError(null);
|
||||
setSessionMissing(false);
|
||||
setLoading(true);
|
||||
resetSessionLoadFailures();
|
||||
void Promise.all([fetchProjects(), fetchSession(), fetchSidebarSessions()]);
|
||||
},
|
||||
}}
|
||||
|
|
|
|||
|
|
@ -634,6 +634,7 @@ function ProjectSidebarInner({
|
|||
{!isDegraded ? (
|
||||
<Link
|
||||
href={projectHref}
|
||||
prefetch={false}
|
||||
onClick={(e) => {
|
||||
e.stopPropagation();
|
||||
onMobileClose?.();
|
||||
|
|
@ -659,6 +660,7 @@ function ProjectSidebarInner({
|
|||
{!isDegraded && orchestratorLink && (
|
||||
<Link
|
||||
href={projectSessionPath(project.id, orchestratorLink.id)}
|
||||
prefetch={false}
|
||||
onClick={(e) => {
|
||||
e.stopPropagation();
|
||||
onMobileClose?.();
|
||||
|
|
|
|||
|
|
@ -20,8 +20,30 @@ function mockVersionResponse(body: {
|
|||
|
||||
describe("UpdateBanner", () => {
|
||||
let fetchMock: ReturnType<typeof vi.fn>;
|
||||
let localStorageStore: Map<string, string>;
|
||||
|
||||
beforeEach(() => {
|
||||
localStorageStore = new Map();
|
||||
Object.defineProperty(window, "localStorage", {
|
||||
configurable: true,
|
||||
writable: true,
|
||||
value: {
|
||||
getItem: (key: string) => localStorageStore.get(key) ?? null,
|
||||
setItem: (key: string, value: string) => {
|
||||
localStorageStore.set(key, value);
|
||||
},
|
||||
removeItem: (key: string) => {
|
||||
localStorageStore.delete(key);
|
||||
},
|
||||
clear: () => {
|
||||
localStorageStore.clear();
|
||||
},
|
||||
get length() {
|
||||
return localStorageStore.size;
|
||||
},
|
||||
key: (index: number) => [...localStorageStore.keys()][index] ?? null,
|
||||
},
|
||||
});
|
||||
window.localStorage.clear();
|
||||
fetchMock = vi.fn();
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ describe("useSessionEvents - mux", () => {
|
|||
await waitFor(() => {
|
||||
expect(fetch).toHaveBeenCalledWith(
|
||||
"/api/sessions?project=proj",
|
||||
expect.objectContaining({ signal: expect.any(AbortSignal), cache: "no-store" }),
|
||||
expect.objectContaining({ cache: "no-store" }),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -0,0 +1,107 @@
|
|||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { __clearInflightFetchesForTest, dedupFetch, fetchJsonWithTimeout } from "../client-fetch";
|
||||
|
||||
describe("dedupFetch", () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
vi.unstubAllGlobals();
|
||||
__clearInflightFetchesForTest();
|
||||
});
|
||||
|
||||
it("shares one underlying request for concurrent requests with the same key", async () => {
|
||||
let resolveFetch: ((response: Response) => void) | undefined;
|
||||
const fetchMock = vi.fn(
|
||||
() =>
|
||||
new Promise<Response>((resolve) => {
|
||||
resolveFetch = resolve;
|
||||
}),
|
||||
);
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
const first = dedupFetch("/api/sessions/ao-187");
|
||||
const second = dedupFetch("/api/sessions/ao-187");
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
|
||||
const response = new Response(JSON.stringify({ ok: true }), { status: 200 });
|
||||
resolveFetch?.(response);
|
||||
|
||||
await expect(first.then((res) => res.json())).resolves.toEqual({ ok: true });
|
||||
await expect(second.then((res) => res.json())).resolves.toEqual({ ok: true });
|
||||
});
|
||||
|
||||
it("starts a new request after the in-flight request settles", async () => {
|
||||
const fetchMock = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(new Response("{}", { status: 200 }))
|
||||
.mockResolvedValueOnce(new Response("{}", { status: 200 }));
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
await dedupFetch("/api/sessions/ao-187");
|
||||
await dedupFetch("/api/sessions/ao-187");
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not coalesce requests with different headers", async () => {
|
||||
const fetchMock = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(new Response("{}", { status: 200 }))
|
||||
.mockResolvedValueOnce(new Response("{}", { status: 200 }));
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
await Promise.all([
|
||||
dedupFetch("/api/sessions/ao-187", { headers: { Accept: "application/json" } }),
|
||||
dedupFetch("/api/sessions/ao-187", { headers: { Accept: "text/plain" } }),
|
||||
]);
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("aborts the underlying fetch when all waiters abort", async () => {
|
||||
const controller = new AbortController();
|
||||
let underlyingSignal: AbortSignal | undefined;
|
||||
const fetchMock = vi.fn(
|
||||
(_input: RequestInfo | URL, init?: RequestInit) =>
|
||||
new Promise<Response>((_resolve, reject) => {
|
||||
underlyingSignal = init?.signal ?? undefined;
|
||||
init?.signal?.addEventListener("abort", () => {
|
||||
reject(new DOMException("The operation was aborted.", "AbortError"));
|
||||
});
|
||||
}),
|
||||
);
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
const request = dedupFetch("/api/sessions/ao-187", { signal: controller.signal });
|
||||
controller.abort();
|
||||
|
||||
await expect(request).rejects.toMatchObject({ name: "AbortError" });
|
||||
expect(underlyingSignal?.aborted).toBe(true);
|
||||
});
|
||||
|
||||
it("times out when response headers arrive but the JSON body stalls", async () => {
|
||||
vi.useFakeTimers();
|
||||
const cancelBody = vi.fn();
|
||||
const body = new ReadableStream({
|
||||
cancel: cancelBody,
|
||||
});
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
new Response(body, {
|
||||
status: 200,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
}),
|
||||
);
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
const request = fetchJsonWithTimeout("/api/sessions/ao-187", {
|
||||
timeoutMs: 100,
|
||||
timeoutMessage: "session timed out",
|
||||
});
|
||||
const expectation = expect(request).rejects.toThrow("session timed out");
|
||||
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
|
||||
await expectation;
|
||||
expect(cancelBody).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
@ -5,6 +5,215 @@ interface FetchJsonOptions extends RequestInit {
|
|||
timeoutMessage?: string;
|
||||
}
|
||||
|
||||
interface InflightFetch {
|
||||
controller: AbortController;
|
||||
consumers: number;
|
||||
maxConsumers: number;
|
||||
promise: Promise<Response>;
|
||||
settled: boolean;
|
||||
}
|
||||
|
||||
const inflightFetches = new Map<string, InflightFetch>();
|
||||
|
||||
function getFetchUrl(input: RequestInfo | URL): string {
|
||||
if (typeof input === "string") return input;
|
||||
if (input instanceof URL) return input.toString();
|
||||
return input.url;
|
||||
}
|
||||
|
||||
function getFetchMethod(input: RequestInfo | URL, init: RequestInit | undefined): string {
|
||||
return (init?.method ?? (input instanceof Request ? input.method : "GET")).toUpperCase();
|
||||
}
|
||||
|
||||
function hashBody(body: BodyInit | null | undefined): string {
|
||||
if (body === null || body === undefined) return "";
|
||||
if (typeof body === "string") return body;
|
||||
if (body instanceof URLSearchParams) return body.toString();
|
||||
if (body instanceof Blob) return `blob:${body.type}:${body.size}`;
|
||||
if (body instanceof ArrayBuffer) return `array-buffer:${body.byteLength}`;
|
||||
if (ArrayBuffer.isView(body)) {
|
||||
return `${body.constructor.name}:${body.byteOffset}:${body.byteLength}`;
|
||||
}
|
||||
if (body instanceof FormData) {
|
||||
return [...body.entries()]
|
||||
.map(([key, value]) =>
|
||||
typeof File !== "undefined" && value instanceof File
|
||||
? `${key}=file:${value.name}:${value.type}:${value.size}`
|
||||
: `${key}=${value}`,
|
||||
)
|
||||
.join("&");
|
||||
}
|
||||
return body.constructor.name;
|
||||
}
|
||||
|
||||
function hashHeaders(input: RequestInfo | URL, init: RequestInit | undefined): string {
|
||||
const headers = new Headers(input instanceof Request ? input.headers : undefined);
|
||||
if (init?.headers) {
|
||||
new Headers(init.headers).forEach((value, key) => {
|
||||
headers.set(key, value);
|
||||
});
|
||||
}
|
||||
return [...headers.entries()]
|
||||
.map(([key, value]) => `${key.toLowerCase()}:${value}`)
|
||||
.sort()
|
||||
.join("\n");
|
||||
}
|
||||
|
||||
function getFetchKey(input: RequestInfo | URL, init: RequestInit | undefined): string {
|
||||
return `${getFetchUrl(input)}|${getFetchMethod(input, init)}|${hashHeaders(input, init)}|${hashBody(init?.body)}`;
|
||||
}
|
||||
|
||||
function cloneResponse(response: Response): Response {
|
||||
if (typeof response.clone !== "function") {
|
||||
throw new TypeError("Cannot clone deduplicated fetch response");
|
||||
}
|
||||
return response.clone();
|
||||
}
|
||||
|
||||
function abortError(): DOMException {
|
||||
return new DOMException("The operation was aborted.", "AbortError");
|
||||
}
|
||||
|
||||
function throwIfAborted(signal: AbortSignal | undefined): void {
|
||||
if (signal?.aborted) {
|
||||
throw abortError();
|
||||
}
|
||||
}
|
||||
|
||||
async function readResponseText(response: Response, signal: AbortSignal | undefined): Promise<string> {
|
||||
throwIfAborted(signal);
|
||||
|
||||
if (!response.body) {
|
||||
return typeof response.text === "function" ? await response.text() : "";
|
||||
}
|
||||
|
||||
const reader = response.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
const chunks: string[] = [];
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
throwIfAborted(signal);
|
||||
let removeAbortListener: () => void = () => {};
|
||||
const abortPromise =
|
||||
signal === undefined
|
||||
? null
|
||||
: new Promise<never>((_, reject) => {
|
||||
const abort = () => {
|
||||
void reader.cancel().catch(() => {});
|
||||
reject(abortError());
|
||||
};
|
||||
if (signal.aborted) {
|
||||
abort();
|
||||
return;
|
||||
}
|
||||
signal.addEventListener("abort", abort, { once: true });
|
||||
removeAbortListener = () => signal.removeEventListener("abort", abort);
|
||||
});
|
||||
|
||||
try {
|
||||
const result = await Promise.race([
|
||||
reader.read(),
|
||||
...(abortPromise ? [abortPromise] : []),
|
||||
]);
|
||||
if (result.done) break;
|
||||
chunks.push(decoder.decode(result.value, { stream: true }));
|
||||
} finally {
|
||||
removeAbortListener();
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
|
||||
chunks.push(decoder.decode());
|
||||
return chunks.join("");
|
||||
}
|
||||
|
||||
async function readResponseJson<T>(
|
||||
response: Response,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<T> {
|
||||
throwIfAborted(signal);
|
||||
if (!response.body && typeof response.json === "function") {
|
||||
return (await response.json()) as T;
|
||||
}
|
||||
return JSON.parse(await readResponseText(response, signal)) as T;
|
||||
}
|
||||
|
||||
export function dedupFetch(input: RequestInfo | URL, init?: RequestInit): Promise<Response> {
|
||||
const key = getFetchKey(input, init);
|
||||
let entry = inflightFetches.get(key);
|
||||
if (!entry) {
|
||||
const controller = new AbortController();
|
||||
const { signal: _signal, ...sharedInit } = init ?? {};
|
||||
const request = fetch(input, { ...sharedInit, signal: controller.signal });
|
||||
const newEntry: InflightFetch = {
|
||||
controller,
|
||||
consumers: 0,
|
||||
maxConsumers: 0,
|
||||
promise: request,
|
||||
settled: false,
|
||||
};
|
||||
request.finally(() => {
|
||||
newEntry.settled = true;
|
||||
inflightFetches.delete(key);
|
||||
}).catch(() => {
|
||||
// The original request promise is returned to callers; this side-effect
|
||||
// chain only prevents an unhandled rejection from the cleanup branch.
|
||||
});
|
||||
entry = newEntry;
|
||||
inflightFetches.set(key, entry);
|
||||
}
|
||||
|
||||
entry.consumers += 1;
|
||||
entry.maxConsumers = Math.max(entry.maxConsumers, entry.consumers);
|
||||
|
||||
const callerSignal = init?.signal ?? undefined;
|
||||
let removeAbortListener: () => void = () => {};
|
||||
let released = false;
|
||||
const release = () => {
|
||||
if (released) return;
|
||||
released = true;
|
||||
removeAbortListener();
|
||||
entry.consumers -= 1;
|
||||
if (entry.consumers === 0 && !entry.settled) {
|
||||
entry.controller.abort();
|
||||
}
|
||||
};
|
||||
|
||||
const responsePromise = entry.promise.then((response) =>
|
||||
entry.maxConsumers > 1 ? cloneResponse(response) : response,
|
||||
);
|
||||
|
||||
const abortPromise =
|
||||
callerSignal === undefined
|
||||
? null
|
||||
: new Promise<never>((_, reject) => {
|
||||
const abort = () => {
|
||||
release();
|
||||
reject(abortError());
|
||||
};
|
||||
if (callerSignal.aborted) {
|
||||
abort();
|
||||
return;
|
||||
}
|
||||
callerSignal.addEventListener("abort", abort, { once: true });
|
||||
removeAbortListener = () => callerSignal.removeEventListener("abort", abort);
|
||||
});
|
||||
|
||||
return Promise.race([responsePromise, ...(abortPromise ? [abortPromise] : [])]).finally(
|
||||
release,
|
||||
);
|
||||
}
|
||||
|
||||
export function __clearInflightFetchesForTest(): void {
|
||||
for (const entry of inflightFetches.values()) {
|
||||
entry.controller.abort();
|
||||
}
|
||||
inflightFetches.clear();
|
||||
}
|
||||
|
||||
function mergeAbortSignals(
|
||||
signals: Array<AbortSignal | null | undefined>,
|
||||
): AbortSignal | undefined {
|
||||
|
|
@ -26,14 +235,23 @@ function mergeAbortSignals(
|
|||
return controller.signal;
|
||||
}
|
||||
|
||||
async function readErrorMessage(response: Response): Promise<string> {
|
||||
async function readErrorMessage(
|
||||
response: Response,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<string> {
|
||||
try {
|
||||
const payload = (await response.json()) as { error?: string; message?: string } | null;
|
||||
const payload = await readResponseJson<{ error?: string; message?: string } | null>(
|
||||
response,
|
||||
signal,
|
||||
);
|
||||
const message = payload?.error ?? payload?.message;
|
||||
if (typeof message === "string" && message.trim().length > 0) {
|
||||
return message.trim();
|
||||
}
|
||||
} catch {
|
||||
} catch (error) {
|
||||
if (signal?.aborted || (error instanceof DOMException && error.name === "AbortError")) {
|
||||
throw error;
|
||||
}
|
||||
// Ignore parse failures and fall back to status text.
|
||||
}
|
||||
|
||||
|
|
@ -52,6 +270,7 @@ export async function fetchJsonWithTimeout<T>(
|
|||
const { timeoutMs = 8_000, timeoutMessage, signal, ...init } = options;
|
||||
const timeoutController = new AbortController();
|
||||
let timer: ReturnType<typeof setTimeout> | null = null;
|
||||
let removeAbortListener: () => void = () => {};
|
||||
let timedOut = false;
|
||||
|
||||
try {
|
||||
|
|
@ -68,9 +287,28 @@ export async function fetchJsonWithTimeout<T>(
|
|||
reject(new Error(timeoutMessage ?? `Request timed out after ${timeoutMs}ms`));
|
||||
}, timeoutMs);
|
||||
});
|
||||
void timeoutPromise.catch(() => {});
|
||||
|
||||
const abortPromise =
|
||||
mergedSignal === undefined
|
||||
? null
|
||||
: new Promise<never>((_, reject) => {
|
||||
const abort = () => {
|
||||
if (!timedOut) {
|
||||
reject(abortError());
|
||||
}
|
||||
};
|
||||
if (mergedSignal.aborted) {
|
||||
abort();
|
||||
return;
|
||||
}
|
||||
mergedSignal.addEventListener("abort", abort, { once: true });
|
||||
removeAbortListener = () => mergedSignal.removeEventListener("abort", abort);
|
||||
});
|
||||
void abortPromise?.catch(() => {});
|
||||
|
||||
const response = await Promise.race([
|
||||
fetch(input, requestInit).catch((error: unknown) => {
|
||||
dedupFetch(input, requestInit).catch((error: unknown) => {
|
||||
if (timedOut) {
|
||||
throw new Error(timeoutMessage ?? `Request timed out after ${timeoutMs}ms`, {
|
||||
cause: error,
|
||||
|
|
@ -79,16 +317,32 @@ export async function fetchJsonWithTimeout<T>(
|
|||
throw error;
|
||||
}),
|
||||
timeoutPromise,
|
||||
...(abortPromise ? [abortPromise] : []),
|
||||
]);
|
||||
|
||||
const readWithTimeout = <T>(read: Promise<T>): Promise<T> =>
|
||||
Promise.race([
|
||||
read.catch((error: unknown) => {
|
||||
if (timedOut) {
|
||||
throw new Error(timeoutMessage ?? `Request timed out after ${timeoutMs}ms`, {
|
||||
cause: error,
|
||||
});
|
||||
}
|
||||
throw error;
|
||||
}),
|
||||
timeoutPromise,
|
||||
...(abortPromise ? [abortPromise] : []),
|
||||
]);
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(await readErrorMessage(response));
|
||||
throw new Error(await readWithTimeout(readErrorMessage(response, mergedSignal)));
|
||||
}
|
||||
|
||||
return (await response.json()) as T;
|
||||
return await readWithTimeout(readResponseJson<T>(response, mergedSignal));
|
||||
} finally {
|
||||
if (timer) {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
removeAbortListener();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue