mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix: retry transient management HTTP requests (#7632)
Retry idempotent coding-agent management requests for transient transport and HTTP failures while keeping one overall timeout budget across attempts. Keep install telemetry single-shot because retrying it has no user-facing impact and could duplicate a side-effecting report.
This commit is contained in:
parent
a80008b96c
commit
46b53b995b
9 changed files with 226 additions and 21 deletions
|
|
@ -1,5 +1,6 @@
|
|||
import type { Api, Model, ModelsStoreEntry, Provider } from "@earendil-works/pi-ai";
|
||||
import { VERSION } from "../config.ts";
|
||||
import { fetchWithRetry } from "../utils/management-http.ts";
|
||||
import { getPiUserAgent } from "../utils/pi-user-agent.ts";
|
||||
|
||||
const DEFAULT_CATALOG_BASE_URL = "https://pi.dev";
|
||||
|
|
@ -77,7 +78,7 @@ export function withRemoteCatalog(
|
|||
// leave the overlay empty.
|
||||
const validator = stored?.models.length ? stored.etag : undefined;
|
||||
const url = new URL(`/api/models/providers/${encodeURIComponent(provider.id)}`, catalogBaseUrl);
|
||||
const response = await fetch(url, {
|
||||
const response = await fetchWithRetry(url, {
|
||||
headers: {
|
||||
accept: "application/json",
|
||||
"User-Agent": getPiUserAgent(VERSION),
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ import { DefaultResourceLoader } from "./core/resource-loader.ts";
|
|||
import { SettingsManager } from "./core/settings-manager.ts";
|
||||
import { hasTrustRequiringProjectResources, ProjectTrustStore } from "./core/trust-manager.ts";
|
||||
import { spawnProcess } from "./utils/child-process.ts";
|
||||
import { getLatestPiRelease, isNewerPackageVersion } from "./utils/version-check.ts";
|
||||
import { formatVersionCheckError, getLatestPiRelease, isNewerPackageVersion } from "./utils/version-check.ts";
|
||||
import {
|
||||
cleanupWindowsSelfUpdateQuarantine,
|
||||
quarantineWindowsNativeDependencies,
|
||||
|
|
@ -476,10 +476,11 @@ interface SelfUpdatePlan {
|
|||
async function getSelfUpdatePlan(force: boolean): Promise<SelfUpdatePlan> {
|
||||
let latestRelease: Awaited<ReturnType<typeof getLatestPiRelease>>;
|
||||
try {
|
||||
latestRelease = await getLatestPiRelease(VERSION);
|
||||
latestRelease = await getLatestPiRelease(VERSION, { retry: true });
|
||||
} catch (error: unknown) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
throw new Error(`Could not determine latest ${APP_NAME} version: ${message}`);
|
||||
throw new Error(`Could not determine latest ${APP_NAME} version: ${formatVersionCheckError(error)}`, {
|
||||
cause: error,
|
||||
});
|
||||
}
|
||||
if (!latestRelease) {
|
||||
throw new Error(`Could not determine latest ${APP_NAME} version.`);
|
||||
|
|
|
|||
66
packages/coding-agent/src/utils/management-http.ts
Normal file
66
packages/coding-agent/src/utils/management-http.ts
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
const RETRYABLE_STATUS_CODES = new Set([408, 425, 429, 500, 502, 503, 504]);
|
||||
|
||||
export interface FetchRetryOptions {
|
||||
/** Number of additional attempts after the initial request. Defaults to two. */
|
||||
maxRetries?: number;
|
||||
/** Retry transient HTTP responses as well as transport failures. Defaults to true. */
|
||||
retryOnStatus?: boolean;
|
||||
/** Per-attempt timeout. A new timeout is created for every attempt. */
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch a management HTTP resource with a bounded immediate retry.
|
||||
*
|
||||
* This is intentionally a transport-level helper for idempotent management
|
||||
* requests (version checks, catalogs, and downloads). It must not be used for
|
||||
* agent/model operations: those can fail after the HTTP request starts and are
|
||||
* retried by their semantic caller instead.
|
||||
*
|
||||
* Caller cancellation is terminal. When timeoutMs is supplied, it is the
|
||||
* overall time budget shared by all attempts.
|
||||
*/
|
||||
export async function fetchWithRetry(
|
||||
input: RequestInfo | URL,
|
||||
init: RequestInit | undefined = undefined,
|
||||
options: FetchRetryOptions = {},
|
||||
): Promise<Response> {
|
||||
const maxRetries =
|
||||
options.maxRetries === undefined || !Number.isFinite(options.maxRetries)
|
||||
? 2
|
||||
: Math.max(0, Math.floor(options.maxRetries));
|
||||
const retryOnStatus = options.retryOnStatus ?? true;
|
||||
const parentSignal = init?.signal;
|
||||
const timeoutSignal =
|
||||
options.timeoutMs !== undefined && options.timeoutMs > 0 ? AbortSignal.timeout(options.timeoutMs) : undefined;
|
||||
const signal = timeoutSignal
|
||||
? parentSignal
|
||||
? AbortSignal.any([parentSignal, timeoutSignal])
|
||||
: timeoutSignal
|
||||
: parentSignal;
|
||||
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
signal?.throwIfAborted();
|
||||
|
||||
try {
|
||||
const response = await fetch(input, signal ? { ...init, signal } : init);
|
||||
const shouldRetry = retryOnStatus && RETRYABLE_STATUS_CODES.has(response.status) && attempt < maxRetries;
|
||||
if (!shouldRetry) return response;
|
||||
try {
|
||||
await response.body?.cancel();
|
||||
} catch {
|
||||
// The response is being discarded before a retry. There is nothing useful to
|
||||
// do if cancelling its body also fails.
|
||||
}
|
||||
} catch (error) {
|
||||
if (
|
||||
parentSignal?.aborted ||
|
||||
timeoutSignal?.aborted ||
|
||||
(error instanceof Error && error.name === "AbortError" && timeoutSignal === undefined) ||
|
||||
attempt >= maxRetries
|
||||
) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -6,6 +6,7 @@ import { join } from "path";
|
|||
import { Readable } from "stream";
|
||||
import { pipeline } from "stream/promises";
|
||||
import { APP_NAME, getBinDir } from "../config.ts";
|
||||
import { fetchWithRetry } from "./management-http.ts";
|
||||
|
||||
const TOOLS_DIR = getBinDir();
|
||||
const NETWORK_TIMEOUT_MS = 10_000;
|
||||
|
|
@ -105,10 +106,13 @@ export function getToolPath(tool: "fd" | "rg"): string | null {
|
|||
|
||||
// Fetch latest release version from GitHub
|
||||
async function getLatestVersion(repo: string): Promise<string> {
|
||||
const response = await fetch(`https://api.github.com/repos/${repo}/releases/latest`, {
|
||||
headers: { "User-Agent": `${APP_NAME}-coding-agent` },
|
||||
signal: AbortSignal.timeout(NETWORK_TIMEOUT_MS),
|
||||
});
|
||||
const response = await fetchWithRetry(
|
||||
`https://api.github.com/repos/${repo}/releases/latest`,
|
||||
{
|
||||
headers: { "User-Agent": `${APP_NAME}-coding-agent` },
|
||||
},
|
||||
{ timeoutMs: NETWORK_TIMEOUT_MS },
|
||||
);
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`GitHub API error: ${response.status}`);
|
||||
|
|
@ -120,9 +124,7 @@ async function getLatestVersion(repo: string): Promise<string> {
|
|||
|
||||
// Download a file from URL
|
||||
async function downloadFile(url: string, dest: string): Promise<void> {
|
||||
const response = await fetch(url, {
|
||||
signal: AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS),
|
||||
});
|
||||
const response = await fetchWithRetry(url, undefined, { timeoutMs: DOWNLOAD_TIMEOUT_MS });
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to download: ${response.status}`);
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import { compare, valid } from "semver";
|
||||
import { fetchWithRetry } from "./management-http.ts";
|
||||
import { getPiUserAgent } from "./pi-user-agent.ts";
|
||||
|
||||
const LATEST_VERSION_URL = "https://pi.dev/api/latest-version";
|
||||
|
|
@ -10,6 +11,24 @@ export interface LatestPiRelease {
|
|||
note?: string;
|
||||
}
|
||||
|
||||
/** Include useful errno details hidden behind Node's generic "fetch failed" error. */
|
||||
export function formatVersionCheckError(error: unknown): string {
|
||||
const rootMessage = error instanceof Error && error.message ? error.message : String(error);
|
||||
const cause = error instanceof Error ? error.cause : undefined;
|
||||
const causes = cause instanceof AggregateError ? cause.errors : cause === undefined ? [] : [cause];
|
||||
const codes = causes
|
||||
.map((value) =>
|
||||
typeof value === "object" && value !== null && "code" in value && typeof value.code === "string"
|
||||
? value.code
|
||||
: undefined,
|
||||
)
|
||||
.filter((code): code is string => code !== undefined);
|
||||
|
||||
if (codes.length > 0) return `${rootMessage} (${[...new Set(codes)].join(", ")})`;
|
||||
const causeMessage = causes.find((value): value is Error => value instanceof Error && value.message)?.message;
|
||||
return causeMessage ? `${rootMessage} (cause: ${causeMessage})` : rootMessage;
|
||||
}
|
||||
|
||||
export function comparePackageVersions(leftVersion: string, rightVersion: string): number | undefined {
|
||||
const left = valid(leftVersion.trim());
|
||||
const right = valid(rightVersion.trim());
|
||||
|
|
@ -29,17 +48,23 @@ export function isNewerPackageVersion(candidateVersion: string, currentVersion:
|
|||
|
||||
export async function getLatestPiRelease(
|
||||
currentVersion: string,
|
||||
options: { timeoutMs?: number } = {},
|
||||
options: { timeoutMs?: number; retry?: boolean } = {},
|
||||
): Promise<LatestPiRelease | undefined> {
|
||||
if (process.env.PI_OFFLINE) return undefined;
|
||||
|
||||
const response = await fetch(LATEST_VERSION_URL, {
|
||||
headers: {
|
||||
"User-Agent": getPiUserAgent(currentVersion),
|
||||
accept: "application/json",
|
||||
const response = await fetchWithRetry(
|
||||
LATEST_VERSION_URL,
|
||||
{
|
||||
headers: {
|
||||
"User-Agent": getPiUserAgent(currentVersion),
|
||||
accept: "application/json",
|
||||
},
|
||||
},
|
||||
signal: AbortSignal.timeout(options.timeoutMs ?? DEFAULT_VERSION_CHECK_TIMEOUT_MS),
|
||||
});
|
||||
{
|
||||
maxRetries: options.retry ? 2 : 0,
|
||||
timeoutMs: options.timeoutMs ?? DEFAULT_VERSION_CHECK_TIMEOUT_MS,
|
||||
},
|
||||
);
|
||||
if (!response.ok) return undefined;
|
||||
|
||||
const data = (await response.json()) as {
|
||||
|
|
@ -62,7 +87,7 @@ export async function getLatestPiRelease(
|
|||
|
||||
export async function getLatestPiVersion(
|
||||
currentVersion: string,
|
||||
options: { timeoutMs?: number } = {},
|
||||
options: { timeoutMs?: number; retry?: boolean } = {},
|
||||
): Promise<string | undefined> {
|
||||
return (await getLatestPiRelease(currentVersion, options))?.version;
|
||||
}
|
||||
|
|
|
|||
52
packages/coding-agent/test/management-http.test.ts
Normal file
52
packages/coding-agent/test/management-http.test.ts
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { fetchWithRetry } from "../src/utils/management-http.ts";
|
||||
|
||||
afterEach(() => vi.restoreAllMocks());
|
||||
|
||||
describe("fetchWithRetry", () => {
|
||||
it("retries a transient transport failure once", async () => {
|
||||
const fetchMock = vi
|
||||
.spyOn(globalThis, "fetch")
|
||||
.mockRejectedValueOnce(new Error("fetch failed"))
|
||||
.mockRejectedValueOnce(new Error("fetch failed"))
|
||||
.mockResolvedValueOnce(Response.json({ ok: true }));
|
||||
|
||||
const response = await fetchWithRetry("https://example.test");
|
||||
expect(response.ok).toBe(true);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("shares the timeout budget across attempts", async () => {
|
||||
const signals: AbortSignal[] = [];
|
||||
const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(async (_input, init) => {
|
||||
if (init?.signal) signals.push(init.signal);
|
||||
if (signals.length === 1) throw new Error("fetch failed");
|
||||
return Response.json({ ok: true });
|
||||
});
|
||||
|
||||
const response = await fetchWithRetry("https://example.test", undefined, { timeoutMs: 1000 });
|
||||
expect(response.ok).toBe(true);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
expect(signals[0]).toBe(signals[1]);
|
||||
});
|
||||
|
||||
it("retries transient HTTP responses and returns the successful response", async () => {
|
||||
const fetchMock = vi
|
||||
.spyOn(globalThis, "fetch")
|
||||
.mockResolvedValueOnce(new Response("busy", { status: 503 }))
|
||||
.mockResolvedValueOnce(Response.json({ ok: true }));
|
||||
|
||||
const response = await fetchWithRetry("https://example.test");
|
||||
expect(response.ok).toBe(true);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not retry caller cancellation", async () => {
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
const fetchMock = vi.spyOn(globalThis, "fetch");
|
||||
|
||||
await expect(fetchWithRetry("https://example.test", { signal: controller.signal })).rejects.toThrow();
|
||||
expect(fetchMock).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
@ -499,6 +499,30 @@ describe("package commands", () => {
|
|||
}
|
||||
});
|
||||
|
||||
it("retries a transient self-update version check", async () => {
|
||||
const previousSkipVersionCheck = process.env.PI_SKIP_VERSION_CHECK;
|
||||
delete process.env.PI_SKIP_VERSION_CHECK;
|
||||
const fetchMock = vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error("fetch failed"))
|
||||
.mockRejectedValueOnce(new Error("fetch failed"))
|
||||
.mockResolvedValueOnce(Response.json({ version: VERSION }));
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
|
||||
|
||||
try {
|
||||
await expect(runPackageCommandDirectly(["update", "--self"])).resolves.toBeUndefined();
|
||||
expect(fetchMock).toHaveBeenCalledTimes(3);
|
||||
expect(errorSpy).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
if (previousSkipVersionCheck === undefined) delete process.env.PI_SKIP_VERSION_CHECK;
|
||||
else process.env.PI_SKIP_VERSION_CHECK = previousSkipVersionCheck;
|
||||
logSpy.mockRestore();
|
||||
errorSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("uses the update check version for forced self updates even when current", async () => {
|
||||
const globalPrefix = join(tempDir, "global-prefix");
|
||||
const projectPrefix = join(tempDir, "project-prefix");
|
||||
|
|
|
|||
|
|
@ -166,6 +166,8 @@ describe("remote catalog provider", () => {
|
|||
headers: { "content-type": "application/json", etag: '"catalog-1"' },
|
||||
}),
|
||||
new Response("rate limited", { status: 429 }),
|
||||
new Response("rate limited", { status: 429 }),
|
||||
new Response("rate limited", { status: 429 }),
|
||||
new Response(null, { status: 304, headers: { etag: '"catalog-1"' } }),
|
||||
];
|
||||
const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => responses.shift() as Response);
|
||||
|
|
@ -180,7 +182,7 @@ describe("remote catalog provider", () => {
|
|||
expect(stored?.models.map((entry) => entry.id)).toEqual(["dynamic"]);
|
||||
|
||||
await refreshProvider(provider, store, { force: true });
|
||||
expect(fetchSpy.mock.calls[2]?.[1]?.headers).toMatchObject({ "if-none-match": '"catalog-1"' });
|
||||
expect(fetchSpy.mock.calls[4]?.[1]?.headers).toMatchObject({ "if-none-match": '"catalog-1"' });
|
||||
expect(provider.getModels().map((entry) => entry.id)).toEqual(["static", "dynamic"]);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
|||
import {
|
||||
checkForNewPiVersion,
|
||||
comparePackageVersions,
|
||||
formatVersionCheckError,
|
||||
getLatestPiRelease,
|
||||
getLatestPiVersion,
|
||||
isNewerPackageVersion,
|
||||
|
|
@ -57,6 +58,37 @@ describe("version checks", () => {
|
|||
);
|
||||
});
|
||||
|
||||
it("retries a transient version request when explicitly requested", async () => {
|
||||
const fetchMock = vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error("fetch failed"))
|
||||
.mockRejectedValueOnce(new Error("fetch failed"))
|
||||
.mockResolvedValueOnce(Response.json({ version: "1.2.4" }));
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
await expect(getLatestPiRelease("1.2.3", { retry: true })).resolves.toEqual({ version: "1.2.4" });
|
||||
expect(fetchMock).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("keeps automatic version checks to one request", async () => {
|
||||
const fetchMock = vi.fn().mockRejectedValue(new Error("fetch failed"));
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
|
||||
await expect(checkForNewPiVersion("1.2.3")).resolves.toBeUndefined();
|
||||
expect(fetchMock).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("formats nested network error details", () => {
|
||||
const error = new Error("fetch failed", {
|
||||
cause: new AggregateError([
|
||||
Object.assign(new Error("connect timeout"), { code: "ETIMEDOUT" }),
|
||||
Object.assign(new Error("network unreachable"), { code: "ENETUNREACH" }),
|
||||
]),
|
||||
});
|
||||
|
||||
expect(formatVersionCheckError(error)).toBe("fetch failed (ETIMEDOUT, ENETUNREACH)");
|
||||
});
|
||||
|
||||
it("returns the active package metadata from the version check api", async () => {
|
||||
const fetchMock = vi.fn(async () =>
|
||||
Response.json({
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue