boos/BoosAPI
2
0
Fork 2
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add proxy first-byte timeout and log badges (#383)

Browse Source 
* add proxy first-byte timeout and log badges

* fix first-byte timeout review feedback
This commit is contained in:
Cita
2026-04-02 10:00:11 +08:00
committed by GitHub
parent bcd758edac
commit 180d17adfc
43 changed files with 4049 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/routes/api/settings.events.test.ts
+27
View File
@@ -65,6 +65,7 @@ describe('settings and auth events', () => {
(config as any).proxyDebugRetentionHours = 24;
(config as any).proxyDebugMaxBodyBytes = 262144;
config.routingFallbackUnitCost = 1;
(config as any).proxyFirstByteTimeoutSec = 0;
(config as any).tokenRouterFailureCooldownMaxSec = 30 * 24 * 60 * 60;
(config as any).disableCrossProtocolFallback = false;
(config as any).telegramEnabled = false;
@@ -508,6 +509,32 @@ describe('settings and auth events', () => {
expect(saved?.value).toBe(JSON.stringify(thirtyDaysSec));
});
it('persists and returns first-byte timeout from runtime settings', async () => {
const updateResponse = await app.inject({
method: 'PUT',
url: '/api/settings/runtime',
payload: {
proxyFirstByteTimeoutSec: 7,
},
});
expect(updateResponse.statusCode).toBe(200);
const updated = updateResponse.json() as { proxyFirstByteTimeoutSec?: number };
expect(updated.proxyFirstByteTimeoutSec).toBe(7);
expect((config as any).proxyFirstByteTimeoutSec).toBe(7);
const saved = await db.select().from(schema.settings).where(eq(schema.settings.key, 'proxy_first_byte_timeout_sec')).get();
expect(saved?.value).toBe(JSON.stringify(7));
const getResponse = await app.inject({
method: 'GET',
url: '/api/settings/runtime',
});
expect(getResponse.statusCode).toBe(200);
const runtime = getResponse.json() as { proxyFirstByteTimeoutSec?: number };
expect(runtime.proxyFirstByteTimeoutSec).toBe(7);
});
it('persists and returns disable cross protocol fallback from runtime settings', async () => {
const updateResponse = await app.inject({
method: 'PUT',
src/server/routes/api/settings.ts
+21
View File
@@ -95,6 +95,7 @@ interface RuntimeSettingsBody {
notifyCooldownSec?: number;
adminIpAllowlist?: string[] | string;
routingFallbackUnitCost?: number;
proxyFirstByteTimeoutSec?: number;
tokenRouterFailureCooldownMaxSec?: number;
routingWeights?: Partial<RoutingWeights>;
proxyErrorKeywords?: string[] | string;
@@ -681,6 +682,12 @@ function applyImportedSettingToRuntime(key: string, value: unknown) {
config.routingFallbackUnitCost = Math.max(1e-6, n);
return;
}
case 'proxy_first_byte_timeout_sec': {
const n = Number(value);
if (!Number.isFinite(n) || n < 0) return;
config.proxyFirstByteTimeoutSec = Math.max(0, Math.trunc(n));
return;
}
case 'token_router_failure_cooldown_max_sec': {
const normalized = normalizeTokenRouterFailureCooldownMaxSec(value);
if (normalized == null) return;
@@ -717,6 +724,7 @@ function getRuntimeSettingsResponse(currentAdminIp = '') {
proxyDebugRetentionHours: config.proxyDebugRetentionHours,
proxyDebugMaxBodyBytes: config.proxyDebugMaxBodyBytes,
routingFallbackUnitCost: config.routingFallbackUnitCost,
proxyFirstByteTimeoutSec: config.proxyFirstByteTimeoutSec,
tokenRouterFailureCooldownMaxSec: config.tokenRouterFailureCooldownMaxSec,
routingWeights: config.routingWeights,
webhookUrl: config.webhookUrl,
@@ -1639,6 +1647,19 @@ export async function settingsRoutes(app: FastifyInstance) {
upsertSetting('routing_fallback_unit_cost', normalized);
}
if (body.proxyFirstByteTimeoutSec !== undefined) {
const nextProxyFirstByteTimeoutSec = Number(body.proxyFirstByteTimeoutSec);
if (!Number.isFinite(nextProxyFirstByteTimeoutSec) || nextProxyFirstByteTimeoutSec < 0) {
return reply.code(400).send({ success: false, message: '首字超时必须是大于等于 0 的数字（秒）' });
}
const normalized = Math.max(0, Math.trunc(nextProxyFirstByteTimeoutSec));
if (normalized !== config.proxyFirstByteTimeoutSec) {
changedLabels.push(`首字超时（${config.proxyFirstByteTimeoutSec}s -> ${normalized}s）`);
}
config.proxyFirstByteTimeoutSec = normalized;
upsertSetting('proxy_first_byte_timeout_sec', normalized);
}
if (body.tokenRouterFailureCooldownMaxSec !== undefined) {
const normalized = normalizeTokenRouterFailureCooldownMaxSec(body.tokenRouterFailureCooldownMaxSec);
if (normalized == null) {
src/server/routes/api/stats.proxy-logs.test.ts
+16
View File
@@ -75,6 +75,8 @@ describe('stats proxy logs routes', () => {
modelRequested: 'gpt-4o',
modelActual: 'gpt-4o',
status: 'success',
isStream: 1,
firstByteLatencyMs: 45,
clientFamily: 'generic',
clientAppId: 'cherry_studio',
clientAppName: 'Cherry Studio',
@@ -92,6 +94,8 @@ describe('stats proxy logs routes', () => {
modelRequested: 'gpt-4o-mini',
modelActual: 'gpt-4o-mini',
status: 'failed',
isStream: 0,
firstByteLatencyMs: 12,
clientFamily: 'codex',
promptTokens: 8,
completionTokens: 2,
@@ -105,6 +109,8 @@ describe('stats proxy logs routes', () => {
modelRequested: 'gpt-4.1',
modelActual: 'gpt-4.1',
status: 'retried',
isStream: 1,
firstByteLatencyMs: 30,
promptTokens: 20,
completionTokens: 4,
totalTokens: 24,
@@ -117,6 +123,8 @@ describe('stats proxy logs routes', () => {
modelRequested: 'claude-3-7-sonnet',
modelActual: 'claude-3-7-sonnet',
status: 'success',
isStream: 1,
firstByteLatencyMs: 88,
promptTokens: 40,
completionTokens: 10,
totalTokens: 50,
@@ -163,6 +171,8 @@ describe('stats proxy logs routes', () => {
expect(body.items[0]?.clientAppId).toBe(null);
expect(body.items[0]?.clientAppName).toBe(null);
expect(body.items[0]?.clientConfidence).toBe(null);
expect(body.items[0]?.isStream).toBe(false);
expect(body.items[0]?.firstByteLatencyMs).toBe(12);
expect(body.items[0]).not.toHaveProperty('billingDetails');
expect(body.clientOptions).toEqual([
{ value: 'family:codex', label: '协议 · Codex' },
@@ -204,6 +214,8 @@ describe('stats proxy logs routes', () => {
modelRequested: 'gpt-5',
modelActual: 'gpt-5',
status: 'success',
isStream: 1,
firstByteLatencyMs: 64,
clientFamily: 'codex',
clientAppId: 'cherry_studio',
clientAppName: 'Cherry Studio',
@@ -238,6 +250,8 @@ describe('stats proxy logs routes', () => {
clientAppId: string | null;
clientAppName: string | null;
clientConfidence: string | null;
isStream: boolean | null;
firstByteLatencyMs: number | null;
billingDetails: Record<string, unknown> | null;
};
@@ -251,6 +265,8 @@ describe('stats proxy logs routes', () => {
expect(body.clientAppId).toBe('cherry_studio');
expect(body.clientAppName).toBe('Cherry Studio');
expect(body.clientConfidence).toBe('exact');
expect(body.isStream).toBe(true);
expect(body.firstByteLatencyMs).toBe(64);
expect(body.billingDetails).toMatchObject({
breakdown: { totalCost: 0.12 },
usage: { promptTokens: 100, completionTokens: 20 },
src/server/routes/api/stats.ts
+4
View File
@@ -497,6 +497,10 @@ function mapProxyLogRow(
const legacyMeta = parseProxyLogMessageMeta(typeof row.proxy_logs.errorMessage === 'string' ? row.proxy_logs.errorMessage : '');
return {
...row.proxy_logs,
isStream: row.proxy_logs.isStream == null ? null : Boolean(row.proxy_logs.isStream),
firstByteLatencyMs: typeof row.proxy_logs.firstByteLatencyMs === 'number'
? row.proxy_logs.firstByteLatencyMs
: null,
...(options?.includeBillingDetails
? { billingDetails: parseProxyLogBillingDetails(row.proxy_logs.billingDetails) }
: {}),
src/server/routes/proxy/completions.ts
+46 -16
View File
@@ -1,5 +1,6 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { config } from '../../config.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js';
@@ -17,6 +18,7 @@ import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { fetchWithObservedFirstByte, getObservedResponseMeta } from '../../proxy-core/firstByteTimeout.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
@@ -46,8 +48,9 @@ export async function completionsProxyRoute(app: FastifyInstance) {
headers: request.headers as Record<string, unknown>,
body,
});
const isStream = body.stream === true;
const isStream = body.stream === true;
const firstByteTimeoutMs = Math.max(0, Math.trunc((config.proxyFirstByteTimeoutSec || 0) * 1000));
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -77,23 +80,37 @@ export async function completionsProxyRoute(app: FastifyInstance) {
const forwardBody = { ...body, model: upstreamModel };
const startTime = Date.now();
try {
const upstream = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const { upstream, firstByteLatencyMs } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const attemptStartedAtMs = Date.now();
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/completions');
const response = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
const response = await fetchWithObservedFirstByte(
async (signal) => fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
signal,
}, getProxyUrlFromExtraConfig(selected.account.extraConfig))),
{
firstByteTimeoutMs,
startedAtMs: attemptStartedAtMs,
},
body: JSON.stringify(forwardBody),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
);
const observedFirstByteLatencyMs = getObservedResponseMeta(response)?.firstByteLatencyMs ?? null;
if (!response.ok) {
const errText = await response.text().catch(() => 'unknown error');
throw new SiteApiEndpointRequestError(errText || 'unknown error', {
status: response.status,
rawErrText: errText || null,
firstByteLatencyMs: observedFirstByteLatencyMs,
});
}
return response;
return {
upstream: response,
firstByteLatencyMs: observedFirstByteLatencyMs,
};
});
if (isStream) {
@@ -191,9 +208,11 @@ export async function completionsProxyRoute(app: FastifyInstance) {
billingDetails,
clientContext,
downstreamPath,
isStream,
firstByteLatencyMs,
);
return;
}
return;
}
const rawText = await upstream.text();
let data: any = rawText;
@@ -228,6 +247,8 @@ export async function completionsProxyRoute(app: FastifyInstance) {
null,
clientContext,
downstreamPath,
isStream,
firstByteLatencyMs,
);
if (shouldRetryProxyRequest(failure.status, errText) && canRetryChannelSelection(retryCount, forcedChannelId)) {
@@ -288,11 +309,14 @@ export async function completionsProxyRoute(app: FastifyInstance) {
billingDetails,
clientContext,
downstreamPath,
isStream,
firstByteLatencyMs,
);
return reply.send(data);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
const firstByteLatencyMs = err instanceof SiteApiEndpointRequestError ? err.firstByteLatencyMs : null;
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status,
errorText,
@@ -314,6 +338,8 @@ export async function completionsProxyRoute(app: FastifyInstance) {
null,
clientContext,
downstreamPath,
isStream,
firstByteLatencyMs,
);
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
@@ -355,6 +381,8 @@ async function logProxy(
billingDetails: unknown = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/completions',
isStream: boolean,
firstByteLatencyMs: number | null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
@@ -375,9 +403,11 @@ async function logProxy(
modelRequested,
modelActual: selected.actualModel || modelRequested,
status,
httpStatus,
latencyMs,
promptTokens,
httpStatus,
isStream,
firstByteLatencyMs,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost,
src/server/routes/proxy/embeddings.ts
+47 -24
View File
@@ -1,5 +1,6 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { config } from '../../config.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js';
@@ -16,6 +17,7 @@ import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { fetchWithObservedFirstByte, getObservedResponseMeta } from '../../proxy-core/firstByteTimeout.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
@@ -40,12 +42,13 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
});
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/embeddings';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const firstByteTimeoutMs = Math.max(0, Math.trunc((config.proxyFirstByteTimeoutSec || 0) * 1000));
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -71,16 +74,25 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
const forwardBody = { ...body, model: upstreamModel };
const startTime = Date.now();
try {
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const { upstream, text, firstByteLatencyMs } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const attemptStartedAtMs = Date.now();
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/embeddings');
const response = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
const response = await fetchWithObservedFirstByte(
async (signal) => fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
signal,
}, getProxyUrlFromExtraConfig(selected.account.extraConfig))),
{
firstByteTimeoutMs,
startedAtMs: attemptStartedAtMs,
},
body: JSON.stringify(forwardBody),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
);
const observedFirstByteLatencyMs = getObservedResponseMeta(response)?.firstByteLatencyMs ?? null;
const status = response.status;
let responseText = '';
try {
@@ -89,6 +101,7 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
if (!response.ok) {
throw new SiteApiEndpointRequestError('unknown error', {
status,
firstByteLatencyMs: observedFirstByteLatencyMs,
cause: error,
});
}
@@ -98,11 +111,13 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status,
rawErrText: responseText || null,
firstByteLatencyMs: observedFirstByteLatencyMs,
});
}
return {
upstream: response,
text: responseText,
firstByteLatencyMs: observedFirstByteLatencyMs,
};
});
@@ -140,11 +155,13 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
logProxy(
selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, clientContext, downstreamPath,
false, firstByteLatencyMs,
);
return reply.code(upstream.status).send(data);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
const firstByteLatencyMs = err instanceof SiteApiEndpointRequestError ? err.firstByteLatencyMs : null;
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status,
errorText,
@@ -166,6 +183,8 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
null,
clientContext,
downstreamPath,
false,
firstByteLatencyMs,
);
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
@@ -206,11 +225,13 @@ async function logProxy(
promptTokens = 0,
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
billingDetails: unknown = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/embeddings',
) {
estimatedCost = 0,
billingDetails: unknown = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/embeddings',
isStream = false,
firstByteLatencyMs: number | null = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
@@ -229,11 +250,13 @@ async function logProxy(
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel || modelRequested,
status,
httpStatus,
latencyMs,
promptTokens,
completionTokens,
status,
httpStatus,
isStream,
firstByteLatencyMs,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost,
billingDetails,
src/server/routes/proxy/endpointFlow.first-byte-timeout.test.ts
+114
View File
@@ -0,0 +1,114 @@
import { describe, expect, it } from 'vitest';
import type { BuiltEndpointRequest } from './endpointFlow.js';
function requestFor(path: string): BuiltEndpointRequest {
return {
endpoint: 'responses',
path,
headers: { 'content-type': 'application/json' },
body: { model: 'gpt-5.2', input: 'hello' },
};
}
function buildDelayedResponse(
bodyText: string,
delayMs: number,
status = 200,
_signal?: AbortSignal,
): Response {
const encoder = new TextEncoder();
const body = new ReadableStream<Uint8Array>({
start(controller) {
setTimeout(() => {
controller.enqueue(encoder.encode(bodyText));
controller.close();
}, delayMs);
},
});
return new Response(body, {
status,
headers: { 'content-type': 'application/json' },
});
}
describe('executeEndpointFlow first-byte timeout', () => {
it('falls through to the next endpoint candidate when the current endpoint times out before any output', async () => {
const { executeEndpointFlow } = await import('./endpointFlow.js');
let timedOutSignal: AbortSignal | undefined;
const dispatchRequest = async (
request: BuiltEndpointRequest,
_targetUrl?: string,
signal?: AbortSignal,
) => (
request.path === '/v1/responses'
? (
timedOutSignal = signal,
buildDelayedResponse(JSON.stringify({ ok: false }), 60, 200, signal)
)
: new Response(JSON.stringify({ ok: true }), {
status: 200,
headers: { 'content-type': 'application/json' },
})
) as unknown as Awaited<ReturnType<typeof import('undici').fetch>>;
const failures: string[] = [];
const result = await executeEndpointFlow({
siteUrl: 'https://example.com',
endpointCandidates: ['responses', 'chat'],
buildRequest: (endpoint: 'responses' | 'chat') => endpoint === 'responses'
? requestFor('/v1/responses')
: { ...requestFor('/v1/chat/completions'), endpoint },
dispatchRequest,
firstByteTimeoutMs: 10,
onAttemptFailure: (ctx: { errText: string }) => {
failures.push(ctx.errText);
},
});
expect(result.ok).toBe(true);
if (result.ok) {
expect(result.upstreamPath).toBe('/v1/chat/completions');
}
expect(timedOutSignal).toBeDefined();
expect(timedOutSignal?.aborted).toBe(true);
expect(failures).toHaveLength(1);
expect(failures[0]).toContain('first byte timeout');
});
it('treats first-byte timeout as terminal when cross-protocol fallback is disabled', async () => {
const { executeEndpointFlow } = await import('./endpointFlow.js');
const attemptedPaths: string[] = [];
const result = await executeEndpointFlow({
siteUrl: 'https://example.com',
endpointCandidates: ['responses', 'chat'],
disableCrossProtocolFallback: true,
buildRequest: (endpoint: 'responses' | 'chat') => endpoint === 'responses'
? requestFor('/v1/responses')
: { ...requestFor('/v1/chat/completions'), endpoint },
dispatchRequest: async (
request: BuiltEndpointRequest,
_targetUrl?: string,
signal?: AbortSignal,
) => {
attemptedPaths.push(request.path);
return (
request.path === '/v1/responses'
? buildDelayedResponse(JSON.stringify({ ok: false }), 60, 200, signal)
: new Response(JSON.stringify({ ok: true }), {
status: 200,
headers: { 'content-type': 'application/json' },
})
) as unknown as Awaited<ReturnType<typeof import('undici').fetch>>;
},
firstByteTimeoutMs: 10,
});
expect(result.ok).toBe(false);
if (!result.ok) {
expect(result.status).toBe(408);
expect(result.errText).toContain('first byte timeout');
}
expect(attemptedPaths).toEqual(['/v1/responses']);
});
});
src/server/routes/proxy/endpointFlow.test.ts
+7 -3
View File
@@ -2,9 +2,13 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
import { fetch } from 'undici';
import type { BuiltEndpointRequest } from './endpointFlow.js';
vi.mock('undici', () => ({
fetch: vi.fn(),
}));
vi.mock('undici', async () => {
const actual = await vi.importActual<typeof import('undici')>('undici');
return {
...actual,
fetch: vi.fn(),
};
});
vi.mock('../../services/siteProxy.js', () => ({
withSiteProxyRequestInit: async (_targetUrl: string, init: RequestInit) => init,
src/server/routes/proxy/images.ts
+84 -12
View File
@@ -1,5 +1,6 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { config } from '../../config.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js';
@@ -15,6 +16,7 @@ import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { fetchWithObservedFirstByte, getObservedResponseMeta } from '../../proxy-core/firstByteTimeout.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
@@ -43,6 +45,7 @@ export async function imagesProxyRoute(app: FastifyInstance) {
headers: request.headers as Record<string, unknown>,
body,
});
const firstByteTimeoutMs = Math.max(0, Math.trunc((config.proxyFirstByteTimeoutSec || 0) * 1000));
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -72,25 +75,37 @@ export async function imagesProxyRoute(app: FastifyInstance) {
const startTime = Date.now();
try {
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const { upstream, text, firstByteLatencyMs } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const attemptStartedAtMs = Date.now();
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/images/generations');
const response = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
const response = await fetchWithObservedFirstByte(
async (signal) => fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
signal,
}, getProxyUrlFromExtraConfig(selected.account.extraConfig))),
{
firstByteTimeoutMs,
startedAtMs: attemptStartedAtMs,
},
body: JSON.stringify(forwardBody),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
);
const observedFirstByteLatencyMs = getObservedResponseMeta(response)?.firstByteLatencyMs ?? null;
const responseText = await response.text();
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status: response.status,
rawErrText: responseText || null,
firstByteLatencyMs: observedFirstByteLatencyMs,
});
}
return {
upstream: response,
text: responseText,
firstByteLatencyMs: observedFirstByteLatencyMs,
};
});
@@ -113,6 +128,8 @@ export async function imagesProxyRoute(app: FastifyInstance) {
0,
downstreamPath,
clientContext,
false,
firstByteLatencyMs,
);
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
@@ -145,11 +162,26 @@ export async function imagesProxyRoute(app: FastifyInstance) {
await recordTokenRouterEventBestEffort('record downstream cost usage', () => (
recordDownstreamCostUsage(request, estimatedCost)
));
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamApiKeyId,
estimatedCost,
downstreamPath,
clientContext,
false,
firstByteLatencyMs,
);
return reply.code(upstream.status).send(data.value);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
const firstByteLatencyMs = err instanceof SiteApiEndpointRequestError ? err.firstByteLatencyMs : null;
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status,
errorText,
@@ -167,6 +199,8 @@ export async function imagesProxyRoute(app: FastifyInstance) {
0,
downstreamPath,
clientContext,
false,
firstByteLatencyMs,
);
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
@@ -216,6 +250,7 @@ export async function imagesProxyRoute(app: FastifyInstance) {
headers: request.headers as Record<string, unknown>,
body: jsonBody || Object.fromEntries(multipartForm?.entries?.() || []),
});
const firstByteTimeoutMs = Math.max(0, Math.trunc((config.proxyFirstByteTimeoutSec || 0) * 1000));
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -244,7 +279,8 @@ export async function imagesProxyRoute(app: FastifyInstance) {
const startTime = Date.now();
try {
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const { upstream, text, firstByteLatencyMs } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const attemptStartedAtMs = Date.now();
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/images/edits');
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
@@ -267,16 +303,29 @@ export async function imagesProxyRoute(app: FastifyInstance) {
model: upstreamModel,
}),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig));
const response = await fetch(targetUrl, requestInit);
const response = await fetchWithObservedFirstByte(
async (signal) => fetch(targetUrl, {
...requestInit,
signal,
}),
{
firstByteTimeoutMs,
startedAtMs: attemptStartedAtMs,
},
);
const observedFirstByteLatencyMs = getObservedResponseMeta(response)?.firstByteLatencyMs ?? null;
const responseText = await response.text();
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status: response.status,
rawErrText: responseText || null,
firstByteLatencyMs: observedFirstByteLatencyMs,
});
}
return {
upstream: response,
text: responseText,
firstByteLatencyMs: observedFirstByteLatencyMs,
};
});
@@ -299,6 +348,8 @@ export async function imagesProxyRoute(app: FastifyInstance) {
0,
downstreamPath,
clientContext,
false,
firstByteLatencyMs,
);
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
@@ -331,11 +382,26 @@ export async function imagesProxyRoute(app: FastifyInstance) {
await recordTokenRouterEventBestEffort('record downstream cost usage', () => (
recordDownstreamCostUsage(request, estimatedCost)
));
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamApiKeyId,
estimatedCost,
downstreamPath,
clientContext,
false,
firstByteLatencyMs,
);
return reply.code(upstream.status).send(data.value);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
const firstByteLatencyMs = err instanceof SiteApiEndpointRequestError ? err.firstByteLatencyMs : null;
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status,
errorText,
@@ -353,6 +419,8 @@ export async function imagesProxyRoute(app: FastifyInstance) {
0,
downstreamPath,
clientContext,
false,
firstByteLatencyMs,
);
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
@@ -402,6 +470,8 @@ async function logProxy(
estimatedCost = 0,
downstreamPath = '/v1/images/generations',
clientContext: DownstreamClientContext | null = null,
isStream = false,
firstByteLatencyMs: number | null = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
@@ -423,6 +493,8 @@ async function logProxy(
modelActual: selected.actualModel || modelRequested,
status,
httpStatus,
isStream,
firstByteLatencyMs,
latencyMs,
promptTokens: 0,
completionTokens: 0,
src/server/routes/proxy/search.ts
+44 -9
View File
@@ -1,5 +1,6 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { config } from '../../config.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js';
@@ -13,6 +14,7 @@ import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { fetchWithObservedFirstByte, getObservedResponseMeta } from '../../proxy-core/firstByteTimeout.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
@@ -71,6 +73,7 @@ export async function searchProxyRoute(app: FastifyInstance) {
headers: request.headers as Record<string, unknown>,
body,
});
const firstByteTimeoutMs = Math.max(0, Math.trunc((config.proxyFirstByteTimeoutSec || 0) * 1000));
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -104,25 +107,37 @@ export async function searchProxyRoute(app: FastifyInstance) {
const startTime = Date.now();
try {
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const { upstream, text, firstByteLatencyMs } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const attemptStartedAtMs = Date.now();
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/search');
const response = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
const response = await fetchWithObservedFirstByte(
async (signal) => fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
signal,
}, getProxyUrlFromExtraConfig(selected.account.extraConfig))),
{
firstByteTimeoutMs,
startedAtMs: attemptStartedAtMs,
},
body: JSON.stringify(forwardBody),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
);
const observedFirstByteLatencyMs = getObservedResponseMeta(response)?.firstByteLatencyMs ?? null;
const responseText = await response.text();
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status: response.status,
rawErrText: responseText || null,
firstByteLatencyMs: observedFirstByteLatencyMs,
});
}
return {
upstream: response,
text: responseText,
firstByteLatencyMs: observedFirstByteLatencyMs,
};
});
@@ -134,11 +149,25 @@ export async function searchProxyRoute(app: FastifyInstance) {
tokenRouter.recordSuccess(selected.channel.id, latency, 0, upstreamModel)
));
recordDownstreamCostUsage(request, 0);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, clientContext, downstreamPath);
logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamApiKeyId,
clientContext,
downstreamPath,
false,
firstByteLatencyMs,
);
return reply.code(upstream.status).send(data);
} catch (error: any) {
const status = error instanceof SiteApiEndpointRequestError ? (error.status || 0) : 0;
const errorText = error?.message || 'network error';
const firstByteLatencyMs = error instanceof SiteApiEndpointRequestError ? error.firstByteLatencyMs : null;
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status,
errorText,
@@ -155,6 +184,8 @@ export async function searchProxyRoute(app: FastifyInstance) {
downstreamApiKeyId,
clientContext,
downstreamPath,
false,
firstByteLatencyMs,
);
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
@@ -194,6 +225,8 @@ async function logProxy(
downstreamApiKeyId: number | null = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/search',
isStream = false,
firstByteLatencyMs: number | null = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
@@ -206,6 +239,8 @@ async function logProxy(
modelActual: selected.actualModel || modelRequested,
status,
httpStatus,
isStream,
firstByteLatencyMs,
latencyMs,
promptTokens: 0,
completionTokens: 0,