feat(proxy): implement metrics persistence, traffic logs DB, and advanced monitoring

- Add SQLite schema for proxy metrics and traffic logs
- Implement ProxyMetricsRepo with historical aggregates
- Add TrafficLogsRepo for DB-backed traffic log queries
- Create proxy-advanced IPC handlers (metrics, traffic logs, circuit breaker, parity)
- Add ProxyMetricsService with connection snapshots and cleanup
- Add ProxyReplayService and ProxyIdeConfigService
- Implement MonitorTab and ToolsTab for advanced proxy features
- Add TrafficLogTable with sorting and filtering
- Fix i18n keys for proxy metrics
- Fix generateIdeConfig type mapping
- Update proxy routes with advanced tab integration
This commit is contained in:
Gemini Nexus Bot committed 2026-04-23 00:57:01 -03:00
1 parent c73537d44c
commit 0be7d7fea2
36 files changed
+4551 -882

No files matched your search

+7 -2
View File
@@ -82,12 +82,17 @@ export function safeApiLogValue(value: unknown, maxLength = 50_000): unknown {
const sanitized = sanitizeObject(value);
if (typeof sanitized === 'string' && sanitized.length > maxLength) {
return sanitized.substring(0, maxLength) + `... [truncated ${sanitized.length - maxLength} chars]`;
return (
sanitized.substring(0, maxLength) + `... [truncated ${sanitized.length - maxLength} chars]`
);
}
const stringified = JSON.stringify(sanitized);
if (stringified.length > maxLength) {
return stringified.substring(0, maxLength) + `... [truncated ${stringified.length - maxLength} chars]`;
return (
stringified.substring(0, maxLength) +
`... [truncated ${stringified.length - maxLength} chars]`
);
}
return sanitized;
@@ -0,0 +1,126 @@
import { Injectable, Logger } from '@nestjs/common';
import { networkInterfaces } from 'os';
import { getServerConfig } from '../../server-config';
import { registerProxyAdvancedService } from '../../../ipc/proxy-advanced/service-registry';
export type SupportedIde = 'cursor' | 'vscode' | 'claude-code' | 'jetbrains' | 'opencide';
export interface IdeConfigResult {
ide: SupportedIde;
format: 'json' | 'shell' | 'env' | 'instructions';
content: Record<string, string> | string;
}
@Injectable()
export class ProxyIdeConfigService {
private readonly logger = new Logger(ProxyIdeConfigService.name);
constructor() {
registerProxyAdvancedService('proxyIdeConfig', this);
}
/**
* Return the first non-internal IPv4 address found on the machine.
* Falls back to '127.0.0.1' if no external interface is available.
*/
getLocalIp(): string {
const interfaces = networkInterfaces();
for (const iface of Object.values(interfaces)) {
if (!iface) {
continue;
}
for (const entry of iface) {
if (entry.family === 'IPv4' && !entry.internal) {
return entry.address;
}
}
}
return '127.0.0.1';
}
/**
* Generate a ready-to-paste IDE configuration for the given IDE.
*/
generateIdeConfig(
ide: SupportedIde,
port: number,
apiKey: string,
localIp: string,
): IdeConfigResult {
const baseUrl = `http://${localIp}:${port}/v1`;
switch (ide) {
case 'cursor': {
return {
ide,
format: 'json',
content: {
OPENAI_API_KEY: apiKey,
OPENAI_BASE_URL: baseUrl,
},
};
}
case 'vscode': {
return {
ide,
format: 'json',
content: {
'openai.url': baseUrl,
'openai.apiKey': apiKey,
},
};
}
case 'claude-code': {
return {
ide,
format: 'shell',
content: `export OPENAI_API_KEY=${apiKey} && export OPENAI_BASE_URL=${baseUrl}`,
};
}
case 'jetbrains': {
return {
ide,
format: 'instructions',
content: [
'JetBrains AI Assistant Setup:',
'1. Open Settings (File > Settings or Ctrl+Alt+S).',
'2. Navigate to Languages & Frameworks > AI Assistant.',
'3. Under "Custom OpenAI Endpoint", set:',
` - Base URL: ${baseUrl}`,
` - API Key: ${apiKey}`,
'4. Click "Apply" and restart the IDE if prompted.',
].join('\n'),
};
}
case 'opencide': {
return {
ide,
format: 'env',
content: `OPENAI_API_KEY=${apiKey}\nOPENAI_BASE_URL=${baseUrl}`,
};
}
default: {
// Exhaustive fallback to satisfy TypeScript
const exhaustiveCheck: never = ide;
this.logger.warn(`Unsupported IDE requested: ${String(exhaustiveCheck)}`);
throw new Error(`Unsupported IDE: ${String(exhaustiveCheck)}`);
}
}
}
/**
* Convenience helper that reads the current server config and local IP,
* then delegates to generateIdeConfig.
*/
generateConfigForCurrentServer(ide: SupportedIde): IdeConfigResult | null {
const config = getServerConfig();
if (!config) {
this.logger.warn('Server config not available; cannot generate IDE config');
return null;
}
const port = config.port;
const apiKey = config.api_key;
const localIp = this.getLocalIp();
return this.generateIdeConfig(ide, port, apiKey, localIp);
}
}
@@ -0,0 +1,111 @@
import { Injectable, OnModuleInit } from '@nestjs/common';
import { registerProxyAdvancedService } from '../../../ipc/proxy-advanced/service-registry';
import { ProxyMetricsRepo } from '../../../ipc/database/proxyMetricsHandler';
interface RequestRecord {
timestamp: number;
duration: number;
status: number;
endpoint: string;
}
const HISTORICAL_WINDOW_MS = 60 * 60 * 1000; // 1 hour
const CONNECTION_SNAPSHOT_INTERVAL_MS = 30_000; // 30 seconds
const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; // 24 hours
const RECORDS_TTL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days
@Injectable()
export class ProxyMetricsService implements OnModuleInit {
private readonly maxRecords = 100;
private records: RequestRecord[] = [];
private activeConnectionsCount = 0;
private tokenEvents: Array<{ timestamp: number; tokens: number }> = [];
constructor() {
registerProxyAdvancedService('proxyMetrics', this);
}
onModuleInit(): void {
this.cleanupOldRecords();
setInterval(() => this.cleanupOldRecords(), CLEANUP_INTERVAL_MS);
this.recordConnectionSnapshot();
setInterval(() => this.recordConnectionSnapshot(), CONNECTION_SNAPSHOT_INTERVAL_MS);
}
incrementActiveConnections(): void {
this.activeConnectionsCount++;
}
decrementActiveConnections(): void {
this.activeConnectionsCount = Math.max(0, this.activeConnectionsCount - 1);
}
recordRequest(record: RequestRecord): void {
this.records.push(record);
if (this.records.length > this.maxRecords) {
this.records.shift();
}
ProxyMetricsRepo.recordRequest({
endpoint: record.endpoint,
statusCode: record.status,
durationMs: record.duration,
timestamp: record.timestamp,
accountId: null,
tokensPrompt: null,
tokensCompletion: null,
errorMessage: null,
});
}
recordTokens(tokens: number): void {
if (tokens <= 0) {
return;
}
this.tokenEvents.push({ timestamp: Date.now(), tokens });
this.trimTokenEvents();
}
getMetrics(): {
totalRequests: number;
requestsPerMinute: number;
avgLatency: number;
errorRate: number;
activeConnections: number;
tokenThroughput: number;
} {
const aggregates = ProxyMetricsRepo.getHistoricalAggregates(HISTORICAL_WINDOW_MS);
this.trimTokenEvents();
const tokenThroughput = this.tokenEvents.reduce((sum, e) => sum + e.tokens, 0);
return {
totalRequests: aggregates.totalRequests,
requestsPerMinute: Math.round(aggregates.requestsPerMinute * 100) / 100,
avgLatency: Math.round(aggregates.avgLatencyMs * 100) / 100,
errorRate: Math.round(aggregates.errorRatePercent * 100) / 100,
activeConnections: Math.round(aggregates.avgActiveConnections * 100) / 100,
tokenThroughput,
};
}
cleanupOldRecords(): void {
const cutoff = Date.now() - RECORDS_TTL_MS;
ProxyMetricsRepo.deleteOldRecords(cutoff);
}
private recordConnectionSnapshot(): void {
ProxyMetricsRepo.recordConnectionSnapshot(this.activeConnectionsCount);
}
private trimTokenEvents(): void {
const cutoff = Date.now() - 60_000;
const index = this.tokenEvents.findIndex((e) => e.timestamp >= cutoff);
if (index > 0) {
this.tokenEvents = this.tokenEvents.slice(index);
} else if (index === -1 && this.tokenEvents.length > 0) {
this.tokenEvents = [];
}
}
}
@@ -0,0 +1,299 @@
import { Injectable, Logger, NotFoundException, BadRequestException, Inject } from '@nestjs/common';
import { registerProxyAdvancedService } from '../../../ipc/proxy-advanced/service-registry';
import { ProxyService } from './proxy.service';
import { v4 as uuidv4 } from 'uuid';
import {
OpenAIChatRequest,
AnthropicChatRequest,
GeminiRequest,
} from './interfaces/request-interfaces';
export interface ReplayEntry {
requestId: string;
timestamp: number;
endpoint: string;
method: string;
body: unknown;
headers: Record<string, unknown>;
accountId?: string;
model?: string;
responseSnapshot?: unknown;
}
@Injectable()
export class ProxyReplayService {
private readonly logger = new Logger(ProxyReplayService.name);
private readonly buffer: ReplayEntry[] = [];
private readonly maxSize = 20;
constructor(
@Inject(ProxyService) private readonly proxyService: ProxyService,
) {
registerProxyAdvancedService('proxyReplay', this);
}
recordRequest(entry: ReplayEntry): void {
if (this.buffer.length >= this.maxSize) {
this.buffer.shift();
}
this.buffer.push(entry);
this.logger.debug(`Recorded request ${entry.requestId} for ${entry.endpoint}`);
}
getRecentRequests(limit?: number): ReplayEntry[] {
const recent = [...this.buffer].reverse();
if (limit && limit > 0) {
return recent.slice(0, limit);
}
return recent;
}
async replayRequest(requestId: string): Promise<{ original: ReplayEntry; newResponse: unknown }> {
const entry = this.buffer.find((e) => e.requestId === requestId);
if (!entry) {
throw new NotFoundException(`Request ${requestId} not found in replay buffer`);
}
this.logger.log(`Replaying request ${requestId} for ${entry.endpoint}`);
const newResponse = await this.executeReplay(entry);
return { original: entry, newResponse };
}
private async executeReplay(entry: ReplayEntry): Promise<unknown> {
switch (entry.endpoint) {
case '/v1/chat/completions':
case '/v1/messages/chat/completions': {
const body = entry.body as OpenAIChatRequest;
return this.proxyService.handleChatCompletions(body, uuidv4());
}
case '/v1/completions': {
const originalBody = entry.body as {
model?: string;
prompt?: string | string[];
max_tokens?: number;
temperature?: number;
top_p?: number;
stream?: boolean;
};
const request: OpenAIChatRequest = {
model: originalBody.model ?? 'gemini-3-flash',
messages: [
{
role: 'user',
content: Array.isArray(originalBody.prompt)
? originalBody.prompt.join('\n')
: (originalBody.prompt ?? ''),
},
],
max_tokens: originalBody.max_tokens,
temperature: originalBody.temperature,
top_p: originalBody.top_p,
stream: originalBody.stream,
};
return this.proxyService.handleChatCompletions(request, uuidv4());
}
case '/v1/messages': {
const body = entry.body as AnthropicChatRequest;
return this.proxyService.handleAnthropicMessages(body);
}
case '/v1/responses': {
const originalBody = entry.body as {
model?: string;
instructions?: string;
input?: unknown;
tools?: OpenAIChatRequest['tools'];
max_output_tokens?: number;
temperature?: number;
top_p?: number;
stream?: boolean;
};
const messages: OpenAIChatRequest['messages'] = [];
if (
typeof originalBody.instructions === 'string' &&
originalBody.instructions.trim().length > 0
) {
messages.push({ role: 'system', content: originalBody.instructions });
}
const input = originalBody.input;
if (typeof input === 'string') {
messages.push({ role: 'user', content: input });
} else if (Array.isArray(input)) {
for (const item of input) {
if (typeof item === 'string') {
messages.push({ role: 'user', content: item });
} else if (item && typeof item === 'object') {
const obj = item as Record<string, unknown>;
if (obj.type === 'message' && typeof obj.content === 'string') {
messages.push({
role: typeof obj.role === 'string' ? obj.role : 'user',
content: obj.content,
});
} else if (obj.type === 'message') {
messages.push({
role: typeof obj.role === 'string' ? obj.role : 'user',
content: JSON.stringify(obj.content),
});
}
}
}
}
if (messages.length === 0) {
messages.push({ role: 'user', content: '' });
}
const request: OpenAIChatRequest = {
model: originalBody.model ?? 'gemini-3-flash',
messages,
tools: originalBody.tools,
max_tokens: originalBody.max_output_tokens,
temperature: originalBody.temperature,
top_p: originalBody.top_p,
stream: originalBody.stream,
};
return this.proxyService.handleChatCompletions(request, uuidv4());
}
case '/v1/images/generations': {
const originalBody = entry.body as {
model?: string;
prompt?: string;
size?: string;
quality?: string;
};
const request: OpenAIChatRequest = {
model: originalBody.model ?? 'gemini-3-pro-image',
messages: [{ role: 'user', content: originalBody.prompt ?? '' }],
stream: false,
size: originalBody.size,
quality: originalBody.quality,
};
return this.proxyService.handleChatCompletions(request, uuidv4());
}
case '/v1/images/edits': {
const originalBody = entry.body as {
model?: string;
prompt?: string;
size?: string;
quality?: string;
image?: string | { data?: string; mimeType?: string };
reference_images?: Array<string | { data?: string; mimeType?: string }>;
mask?: string | { data?: string; mimeType?: string };
};
const imageParts = this.collectImageContentParts([
originalBody.image,
originalBody.mask,
...(originalBody.reference_images ?? []),
]);
const request: OpenAIChatRequest = {
model: originalBody.model ?? 'gemini-3-pro-image',
messages: [
{
role: 'user',
content:
imageParts.length > 0
? [
{
type: 'text' as const,
text:
originalBody.prompt ??
'Please edit this image based on the provided instruction.',
},
...imageParts,
]
: (originalBody.prompt ??
'Please edit this image based on the provided instruction.'),
},
],
stream: false,
size: originalBody.size,
quality: originalBody.quality,
};
return this.proxyService.handleChatCompletions(request, uuidv4());
}
case '/v1/audio/transcriptions': {
const originalBody = entry.body as {
model?: string;
prompt?: string;
file?: string | { data?: string; mimeType?: string };
audio?: string | { data?: string; mimeType?: string };
};
const inlineAudio = this.resolveInlineData(originalBody.file ?? originalBody.audio);
if (!inlineAudio) {
throw new BadRequestException("Missing 'file' or 'audio' input for replay");
}
return this.proxyService.handleGeminiGenerateContent(
originalBody.model ?? 'gemini-3-flash',
{
contents: [
{
role: 'user',
parts: [
{
text:
originalBody.prompt ??
'Please transcribe the provided speech audio accurately.',
},
{ inlineData: inlineAudio },
],
},
],
} as GeminiRequest,
);
}
default:
throw new BadRequestException(`Replay not supported for endpoint ${entry.endpoint}`);
}
}
private collectImageContentParts(
entries: Array<string | { data?: string; mimeType?: string } | undefined>,
): Array<{ type: 'image_url'; image_url: { url: string } }> {
const parts: Array<{ type: 'image_url'; image_url: { url: string } }> = [];
for (const entry of entries) {
const inlineData = this.resolveInlineData(entry);
if (!inlineData) {
continue;
}
parts.push({
type: 'image_url',
image_url: {
url: `data:${inlineData.mimeType};base64,${inlineData.data}`,
},
});
}
return parts;
}
private resolveInlineData(input: unknown): { mimeType: string; data: string } | null {
if (!input) {
return null;
}
if (typeof input === 'string') {
const dataUri = input.match(/^data:(?<mime>[^;]+);base64,(?<data>[A-Za-z0-9+/=]+)$/);
if (dataUri?.groups?.mime && dataUri.groups.data) {
return {
mimeType: dataUri.groups.mime,
data: dataUri.groups.data,
};
}
const cleaned = input.replace(/\s+/g, '');
if (cleaned.length > 0) {
return {
mimeType: 'audio/mpeg',
data: cleaned,
};
}
return null;
}
const inputRecord = input as Record<string, unknown>;
const data = typeof inputRecord.data === 'string' ? inputRecord.data : undefined;
if (!data) {
return null;
}
return {
mimeType: typeof inputRecord.mimeType === 'string' ? inputRecord.mimeType : 'audio/mpeg',
data,
};
}
}
+362 -17
View File
@@ -15,6 +15,7 @@ import { FastifyReply, FastifyRequest } from 'fastify';
import { isEmpty, isFunction, isNil, isObjectLike, isPlainObject, isString } from 'lodash-es';
import { v4 as uuidv4 } from 'uuid';
import { ProxyService } from './proxy.service';
import { ProxyReplayService } from './proxy-replay.service';
import { Observable } from 'rxjs';
import { trafficLogger } from '../../../utils/traffic-logger';
import {
@@ -33,6 +34,7 @@ import {
} from '../../../lib/geminiNexus/ModelMapping';
import { getServerConfig } from '../../server-config';
import { TokenManagerService } from './token-manager.service';
import { ProxyMetricsService } from './proxy-metrics.service';
@Controller('v1')
@UseGuards(ProxyGuard)
@@ -41,11 +43,15 @@ export class ProxyController {
constructor(
@Inject(ProxyService) private readonly proxyService: ProxyService,
@Optional() @Inject(ProxyReplayService) private readonly proxyReplayService?: ProxyReplayService,
@Optional() @Inject(TokenManagerService) private readonly tokenManager?: TokenManagerService,
@Optional() @Inject(ProxyMetricsService) private readonly proxyMetrics?: ProxyMetricsService,
) {}
@Get('models')
listModels(@Res() res: FastifyReply) {
this.proxyMetrics?.incrementActiveConnections();
const startTime = Date.now();
try {
const config = getServerConfig();
const customMapping = config?.custom_mapping ?? {};
@@ -65,6 +71,12 @@ export class ProxyController {
object: 'list',
data,
});
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status: HttpStatus.OK,
endpoint: '/v1/models',
});
} catch (error) {
const message = error instanceof Error ? error.message : 'Failed to list models';
this.logger.error(message, error instanceof Error ? error.stack : undefined);
@@ -80,6 +92,14 @@ export class ProxyController {
res.status(status).send({
error: errorPayload,
});
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status,
endpoint: '/v1/models',
});
} finally {
this.proxyMetrics?.decrementActiveConnections();
}
}
@@ -89,6 +109,15 @@ export class ProxyController {
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/chat/completions',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
await this.respondOpenAIChatCompletions(body, res, req);
}
@@ -103,6 +132,15 @@ export class ProxyController {
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/messages/chat/completions',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
await this.respondOpenAIChatCompletions(body, res, req);
}
@@ -117,8 +155,20 @@ export class ProxyController {
top_p?: number;
stream?: boolean;
},
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/completions',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
this.proxyMetrics?.incrementActiveConnections();
const startTime = Date.now();
const request: OpenAIChatRequest = {
model: body.model ?? 'gemini-3-flash',
messages: [
@@ -132,17 +182,41 @@ export class ProxyController {
top_p: body.top_p,
stream: body.stream,
};
let isStream = false;
try {
const result = await this.proxyService.handleChatCompletions(request);
if (body.stream && this.isObservableLike(result)) {
this.writeSseResponse(res, result);
isStream = !!(body.stream && this.isObservableLike(result));
if (isStream) {
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/completions', startTime);
return;
}
const response = result as OpenAIChatResponse;
const duration = Date.now() - startTime;
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status: HttpStatus.OK,
endpoint: '/v1/completions',
});
if (response.usage?.total_tokens) {
this.proxyMetrics?.recordTokens(response.usage.total_tokens);
}
res.status(HttpStatus.OK).send(this.toLegacyTextCompletionsResponse(response));
} catch (error) {
const duration = Date.now() - startTime;
const status = this.resolveErrorHttpStatus(error instanceof Error ? error.message : '');
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status,
endpoint: '/v1/completions',
});
this.sendOpenAIErrorResponse(res, '/v1/completions', error);
} finally {
if (!isStream) {
this.proxyMetrics?.decrementActiveConnections();
}
}
}
@@ -159,21 +233,57 @@ export class ProxyController {
top_p?: number;
stream?: boolean;
},
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/responses',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
this.proxyMetrics?.incrementActiveConnections();
const startTime = Date.now();
const request = this.buildResponsesChatRequest(body);
let isStream = false;
try {
const result = await this.proxyService.handleChatCompletions(request);
if (body.stream && this.isObservableLike(result)) {
this.writeSseResponse(res, result);
isStream = !!(body.stream && this.isObservableLike(result));
if (isStream) {
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/responses', startTime);
return;
}
const response = result as OpenAIChatResponse;
const duration = Date.now() - startTime;
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status: HttpStatus.OK,
endpoint: '/v1/responses',
});
if (response.usage?.total_tokens) {
this.proxyMetrics?.recordTokens(response.usage.total_tokens);
}
res.status(HttpStatus.OK).send(this.toLegacyTextCompletionsResponse(response));
} catch (error) {
const duration = Date.now() - startTime;
const status = this.resolveErrorHttpStatus(error instanceof Error ? error.message : '');
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status,
endpoint: '/v1/responses',
});
this.sendOpenAIErrorResponse(res, '/v1/responses', error);
} finally {
if (!isStream) {
this.proxyMetrics?.decrementActiveConnections();
}
}
}
@@ -186,8 +296,18 @@ export class ProxyController {
size?: string;
quality?: string;
},
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/images/generations',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
const request: OpenAIChatRequest = {
model: body.model ?? 'gemini-3-pro-image',
messages: [
@@ -219,6 +339,15 @@ export class ProxyController {
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/images/edits',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
if (!this.hasMultipartBoundary(req)) {
res
.status(HttpStatus.BAD_REQUEST)
@@ -270,7 +399,26 @@ export class ProxyController {
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyMetrics?.incrementActiveConnections();
const startTime = Date.now();
const endpoint = '/v1/audio/transcriptions';
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/audio/transcriptions',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
if (!this.hasMultipartBoundary(req)) {
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status: HttpStatus.BAD_REQUEST,
endpoint,
});
this.proxyMetrics?.decrementActiveConnections();
res
.status(HttpStatus.BAD_REQUEST)
.send('Invalid `boundary` for `multipart/form-data` request');
@@ -279,6 +427,13 @@ export class ProxyController {
const inlineAudio = this.resolveInlineData(body.file ?? body.audio);
if (!inlineAudio) {
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status: HttpStatus.BAD_REQUEST,
endpoint,
});
this.proxyMetrics?.decrementActiveConnections();
res.status(HttpStatus.BAD_REQUEST).send({
error: {
message: "Missing 'file' or 'audio' input. Provide base64 content or a data URL.",
@@ -313,11 +468,31 @@ export class ProxyController {
.join('')
.trim();
const duration = Date.now() - startTime;
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status: HttpStatus.OK,
endpoint,
});
if (result.usageMetadata?.totalTokenCount) {
this.proxyMetrics?.recordTokens(result.usageMetadata.totalTokenCount);
}
res.status(HttpStatus.OK).send({
text: text ?? '',
});
} catch (error) {
this.sendOpenAIErrorResponse(res, '/v1/audio/transcriptions', error);
const duration = Date.now() - startTime;
const status = this.resolveErrorHttpStatus(error instanceof Error ? error.message : '');
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status,
endpoint,
});
this.sendOpenAIErrorResponse(res, endpoint, error);
} finally {
this.proxyMetrics?.decrementActiveConnections();
}
}
@@ -326,9 +501,12 @@ export class ProxyController {
res: FastifyReply,
req?: FastifyRequest,
) {
this.proxyMetrics?.incrementActiveConnections();
const requestId = uuidv4();
const startTime = Date.now();
trafficLogger.logInbound(requestId, '/v1/chat/completions', body, undefined, { stream: body.stream });
trafficLogger.logInbound(requestId, '/v1/chat/completions', body, undefined, {
stream: body.stream,
});
// Extract x-session-affinity header so OpenCode session stickiness works even when
// the request body does not carry an explicit session_id field.
@@ -336,36 +514,106 @@ export class ProxyController {
? (req.headers['x-session-affinity'] as string | undefined)
: undefined;
let isStream = false;
try {
const result = await this.proxyService.handleChatCompletions(body, requestId, headerSessionKey);
const result = await this.proxyService.handleChatCompletions(
body,
requestId,
headerSessionKey,
);
if (body.stream && this.isObservableLike(result)) {
this.writeSseResponse(res, result, requestId, '/v1/chat/completions', startTime);
isStream = !!(body.stream && this.isObservableLike(result));
if (isStream) {
this.writeSseResponse(res, result as Observable<unknown>, requestId, '/v1/chat/completions', startTime);
return;
} else {
const duration = Date.now() - startTime;
trafficLogger.logOutbound(requestId, '/v1/chat/completions', HttpStatus.OK, result, duration);
trafficLogger.logOutbound(
requestId,
'/v1/chat/completions',
HttpStatus.OK,
result,
duration,
);
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status: HttpStatus.OK,
endpoint: '/v1/chat/completions',
});
const response = result as OpenAIChatResponse;
if (response.usage?.total_tokens) {
this.proxyMetrics?.recordTokens(response.usage.total_tokens);
}
res.status(HttpStatus.OK).send(result);
}
} catch (error) {
trafficLogger.logError(requestId, '/v1/chat/completions', error);
const duration = Date.now() - startTime;
const status = this.resolveErrorHttpStatus(error instanceof Error ? error.message : '');
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status,
endpoint: '/v1/chat/completions',
});
this.sendOpenAIErrorResponse(res, '/v1/chat/completions', error);
} finally {
if (!isStream) {
this.proxyMetrics?.decrementActiveConnections();
}
}
}
@Post('messages')
async anthropicMessages(@Body() body: AnthropicChatRequest, @Res() res: FastifyReply) {
async anthropicMessages(
@Body() body: AnthropicChatRequest,
@Req() req: FastifyRequest,
@Res() res: FastifyReply,
) {
this.proxyReplayService?.recordRequest({
requestId: uuidv4(),
timestamp: Date.now(),
endpoint: '/v1/messages',
method: 'POST',
body,
headers: req.headers as Record<string, unknown>,
model: body.model,
});
this.proxyMetrics?.incrementActiveConnections();
const startTime = Date.now();
let isStream = false;
try {
const result = await this.proxyService.handleAnthropicMessages(body);
isStream = !!(body.stream && this.isObservableLike(result));
if (body.stream && this.isObservableLike(result)) {
this.writeSseResponse(res, result);
if (isStream) {
this.writeSseResponse(res, result as Observable<unknown>, undefined, '/v1/messages', startTime);
return;
} else {
const duration = Date.now() - startTime;
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status: HttpStatus.OK,
endpoint: '/v1/messages',
});
res.status(HttpStatus.OK).send(result);
}
} catch (error) {
const duration = Date.now() - startTime;
const status = this.resolveErrorHttpStatus(error instanceof Error ? error.message : '');
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration,
status,
endpoint: '/v1/messages',
});
this.sendAnthropicErrorResponse(res, '/v1/messages', error);
} finally {
if (!isStream) {
this.proxyMetrics?.decrementActiveConnections();
}
}
}
@@ -762,6 +1010,35 @@ export class ProxyController {
return isObjectLike(value) && isFunction((value as { subscribe?: unknown }).subscribe);
}
private extractAndRecordTokens(payload: string): void {
try {
const lines = payload.split('\n');
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed.startsWith('data: ')) {
continue;
}
const jsonStr = trimmed.slice(6).trim();
if (jsonStr === '[DONE]') {
continue;
}
const parsed = JSON.parse(jsonStr);
const usageMetadata = parsed?.usageMetadata;
if (usageMetadata && typeof usageMetadata.totalTokenCount === 'number') {
this.proxyMetrics?.recordTokens(usageMetadata.totalTokenCount);
return;
}
const usage = parsed?.usage;
if (usage && typeof usage.total_tokens === 'number') {
this.proxyMetrics?.recordTokens(usage.total_tokens);
return;
}
}
} catch {
// Ignore parse errors for token extraction
}
}
private writeSseResponse(
res: FastifyReply,
stream: Observable<unknown>,
@@ -774,6 +1051,7 @@ export class ProxyController {
res.header('Cache-Control', 'no-cache');
res.header('Connection', 'keep-alive');
res.send(stream);
this.proxyMetrics?.decrementActiveConnections();
return;
}
@@ -791,6 +1069,23 @@ export class ProxyController {
res.raw.flushHeaders();
}
let finished = false;
const finish = (status: number) => {
if (finished) {
return;
}
finished = true;
if (startTime && endpoint) {
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status,
endpoint,
});
}
this.proxyMetrics?.decrementActiveConnections();
};
const subscription = stream.subscribe({
next: (chunk) => {
if (res.raw.writableEnded) {
@@ -801,6 +1096,7 @@ export class ProxyController {
if (requestId && endpoint) {
trafficLogger.logSseChunk(requestId, endpoint, chunk);
}
this.extractAndRecordTokens(payload);
if (isFunction((res.raw as any).flush)) {
(res.raw as any).flush();
}
@@ -826,16 +1122,19 @@ export class ProxyController {
\n\n`,
);
res.raw.end();
finish(status);
},
complete: () => {
if (!res.raw.writableEnded) {
res.raw.end();
}
finish(HttpStatus.OK);
},
});
res.raw.on('close', () => {
subscription.unsubscribe();
finish(HttpStatus.OK);
});
}
@@ -844,14 +1143,23 @@ export class ProxyController {
prompt: string,
res: FastifyReply,
): Promise<void> {
this.proxyMetrics?.incrementActiveConnections();
const startTime = Date.now();
const endpoint = '/v1/images/generations';
try {
const result = await this.proxyService.handleChatCompletions(request);
if (result instanceof Observable) {
this.logProxyEndpointError(
'/v1/images/generations',
endpoint,
HttpStatus.INTERNAL_SERVER_ERROR,
'Streaming image generation is not supported by this endpoint',
);
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status: HttpStatus.INTERNAL_SERVER_ERROR,
endpoint,
});
res.status(HttpStatus.INTERNAL_SERVER_ERROR).send({
error: {
message: 'Streaming image generation is not supported by this endpoint',
@@ -865,10 +1173,16 @@ export class ProxyController {
const image = this.extractInlineBase64Image(isString(content) ? content : '');
if (!image) {
this.logProxyEndpointError(
'/v1/images/generations',
endpoint,
HttpStatus.BAD_GATEWAY,
'Upstream did not return inline image data',
);
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status: HttpStatus.BAD_GATEWAY,
endpoint,
});
res.status(HttpStatus.BAD_GATEWAY).send({
error: {
message: 'Upstream did not return inline image data',
@@ -878,6 +1192,15 @@ export class ProxyController {
return;
}
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status: HttpStatus.OK,
endpoint,
});
if (result.usage?.total_tokens) {
this.proxyMetrics?.recordTokens(result.usage.total_tokens);
}
res.status(HttpStatus.OK).send({
created: Math.floor(Date.now() / 1000),
data: [
@@ -898,6 +1221,15 @@ export class ProxyController {
);
const fallbackImage = this.extractInlineBase64ImageFromGeminiResponse(geminiResult);
if (fallbackImage) {
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status: HttpStatus.OK,
endpoint,
});
if (geminiResult.usageMetadata?.totalTokenCount) {
this.proxyMetrics?.recordTokens(geminiResult.usageMetadata.totalTokenCount);
}
res.status(HttpStatus.OK).send({
created: Math.floor(Date.now() / 1000),
data: [
@@ -914,7 +1246,16 @@ export class ProxyController {
}
}
this.sendOpenAIErrorResponse(res, '/v1/images/generations', error, message);
const status = this.resolveErrorHttpStatus(message);
this.proxyMetrics?.recordRequest({
timestamp: startTime,
duration: Date.now() - startTime,
status,
endpoint,
});
this.sendOpenAIErrorResponse(res, endpoint, error, message);
} finally {
this.proxyMetrics?.decrementActiveConnections();
}
}
@@ -1132,7 +1473,11 @@ export class ProxyController {
if (status === HttpStatus.SERVICE_UNAVAILABLE) {
return { type: 'server_error', code: 'temporarily_unavailable' };
}
if (lowered.includes('invalid') || lowered.includes('malformed') || lowered.includes('bad request')) {
if (
lowered.includes('invalid') ||
lowered.includes('malformed') ||
lowered.includes('bad request')
) {
return { type: 'invalid_request_error', code: 'invalid_request_error' };
}
return { type: 'server_error', code: 'internal_error' };
+4 -1
View File
@@ -6,11 +6,14 @@ import { GeminiClient } from './clients/gemini.client';
import { GeminiController } from './gemini.controller';
import { ProxyGuard } from './proxy.guard';
import { TokenUsageService } from '../usage/token-usage.service';
import { ProxyReplayService } from './proxy-replay.service';
import { ProxyMetricsService } from './proxy-metrics.service';
import { ProxyIdeConfigService } from './proxy-ide-config.service';
@Module({
imports: [],
controllers: [ProxyController, GeminiController],
providers: [ProxyService, TokenManagerService, GeminiClient, ProxyGuard, TokenUsageService],
providers: [ProxyService, TokenManagerService, GeminiClient, ProxyGuard, TokenUsageService, ProxyReplayService, ProxyMetricsService, ProxyIdeConfigService],
exports: [TokenManagerService],
})
export class ProxyModule {}
+115 -38
View File
@@ -50,7 +50,10 @@ export class ProxyService {
private async recordUsage(
accountId: string,
model: string,
usage: { promptTokenCount?: number; candidatesTokenCount?: number; totalTokenCount?: number } | undefined | null,
usage:
| { promptTokenCount?: number; candidatesTokenCount?: number; totalTokenCount?: number }
| undefined
| null,
requestType: string,
isEstimated?: boolean,
): Promise<void> {
@@ -79,12 +82,15 @@ export class ProxyService {
private extractUsageFromMetadata(
meta: unknown,
): { promptTokenCount?: number; candidatesTokenCount?: number; totalTokenCount?: number } | undefined {
):
| { promptTokenCount?: number; candidatesTokenCount?: number; totalTokenCount?: number }
| undefined {
if (!meta || typeof meta !== 'object') return undefined;
const m = meta as Record<string, unknown>;
return {
promptTokenCount: typeof m.promptTokenCount === 'number' ? m.promptTokenCount : undefined,
candidatesTokenCount: typeof m.candidatesTokenCount === 'number' ? m.candidatesTokenCount : undefined,
candidatesTokenCount:
typeof m.candidatesTokenCount === 'number' ? m.candidatesTokenCount : undefined,
totalTokenCount: typeof m.totalTokenCount === 'number' ? m.totalTokenCount : undefined,
};
}
@@ -332,13 +338,17 @@ export class ProxyService {
);
} else if (accumulatedText) {
const estimated = this.estimateTokens(accumulatedText);
this.recordUsage(accountId, _model, {
promptTokenCount: 0,
candidatesTokenCount: estimated,
totalTokenCount: estimated,
}, 'anthropic', true).catch((err) =>
this.logger.error('Usage recording failed', err),
);
this.recordUsage(
accountId,
_model,
{
promptTokenCount: 0,
candidatesTokenCount: estimated,
totalTokenCount: estimated,
},
'anthropic',
true,
).catch((err) => this.logger.error('Usage recording failed', err));
}
subscriber.complete();
});
@@ -504,7 +514,7 @@ export class ProxyService {
token.token.upstream_proxy_url,
extraHeaders,
);
return this.passthroughSseStream(stream, token.id, effectiveTargetModel);
return this.passthroughSseStream(stream, token.id, effectiveTargetModel);
} catch (err) {
if (err instanceof Error && this.isProjectContextError(err.message)) {
this.logger.warn(
@@ -526,7 +536,7 @@ export class ProxyService {
token.token.upstream_proxy_url,
extraHeaders,
);
return this.passthroughSseStream(stream, token.id, effectiveTargetModel);
return this.passthroughSseStream(stream, token.id, effectiveTargetModel);
} catch (fallbackErr) {
lastError = fallbackErr;
}
@@ -579,7 +589,9 @@ export class ProxyService {
if (typeof text === 'string') {
accumulatedText += text;
}
} catch { /* ignore */ }
} catch {
/* ignore */
}
}
subscriber.next(chunk.toString());
@@ -597,13 +609,17 @@ export class ProxyService {
);
} else if (accumulatedText) {
const estimated = this.estimateTokens(accumulatedText);
this.recordUsage(accountId, model, {
promptTokenCount: 0,
candidatesTokenCount: estimated,
totalTokenCount: estimated,
}, 'gemini', true).catch((err) =>
this.logger.error('Usage recording failed', err),
);
this.recordUsage(
accountId,
model,
{
promptTokenCount: 0,
candidatesTokenCount: estimated,
totalTokenCount: estimated,
},
'gemini',
true,
).catch((err) => this.logger.error('Usage recording failed', err));
}
subscriber.complete();
});
@@ -859,14 +875,26 @@ export class ProxyService {
// Use v1internal API (same as Anthropic handler)
if (request.stream) {
try {
trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', geminiBody, extraHeaders, { model: effectiveTargetModel });
trafficLogger.logUpstream(
requestId || 'unknown',
'/v1/chat/completions',
geminiBody,
extraHeaders,
{ model: effectiveTargetModel },
);
const stream = await this.geminiClient.streamGenerateInternal(
geminiBody,
token.token.access_token,
token.token.upstream_proxy_url,
extraHeaders,
);
return this.processStreamResponse(stream, request.model, requestId, token.id, effectiveTargetModel);
return this.processStreamResponse(
stream,
request.model,
requestId,
token.id,
effectiveTargetModel,
);
} catch (streamError) {
this.logger.warn(
`Stream path failed for model=${request.model}; falling back to non-stream generation: ${
@@ -874,14 +902,26 @@ export class ProxyService {
}`,
);
trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', geminiBody, extraHeaders, { model: effectiveTargetModel });
trafficLogger.logUpstream(
requestId || 'unknown',
'/v1/chat/completions',
geminiBody,
extraHeaders,
{ model: effectiveTargetModel },
);
const response = await this.generateInternalWithStreamFallback(
geminiBody,
token.token.access_token,
token.token.upstream_proxy_url,
extraHeaders,
);
trafficLogger.logUpstreamResponse(requestId || 'unknown', '/v1/chat/completions', 200, response, 0);
trafficLogger.logUpstreamResponse(
requestId || 'unknown',
'/v1/chat/completions',
200,
response,
0,
);
this.recordUsage(token.id, effectiveTargetModel, response.usageMetadata, 'openai');
this.logger.log(
`Upstream response snippet after stream fallback: ${JSON.stringify(response).substring(0, 500)}`,
@@ -923,24 +963,48 @@ export class ProxyService {
fallbackBody.model = effectiveTargetModel;
this.applyInternalGenerationConstraints(fallbackBody, effectiveTargetModel, token.id);
if (request.stream) {
trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', fallbackBody, extraHeaders, { model: effectiveTargetModel });
trafficLogger.logUpstream(
requestId || 'unknown',
'/v1/chat/completions',
fallbackBody,
extraHeaders,
{ model: effectiveTargetModel },
);
const stream = await this.geminiClient.streamGenerateInternal(
fallbackBody,
token.token.access_token,
token.token.upstream_proxy_url,
extraHeaders,
);
return this.processStreamResponse(stream, request.model, requestId, token.id, effectiveTargetModel);
return this.processStreamResponse(
stream,
request.model,
requestId,
token.id,
effectiveTargetModel,
);
}
trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', fallbackBody, extraHeaders, { model: effectiveTargetModel });
trafficLogger.logUpstream(
requestId || 'unknown',
'/v1/chat/completions',
fallbackBody,
extraHeaders,
{ model: effectiveTargetModel },
);
const response = await this.generateInternalWithStreamFallback(
fallbackBody,
token.token.access_token,
token.token.upstream_proxy_url,
extraHeaders,
);
trafficLogger.logUpstreamResponse(requestId || 'unknown', '/v1/chat/completions', 200, response, 0);
trafficLogger.logUpstreamResponse(
requestId || 'unknown',
'/v1/chat/completions',
200,
response,
0,
);
this.recordUsage(token.id, effectiveTargetModel, response.usageMetadata, 'openai');
const claudeResponse = transformResponse(response);
return this.convertClaudeToOpenAIResponse(claudeResponse, request.model);
@@ -1135,7 +1199,14 @@ export class ProxyService {
try {
const json = JSON.parse(dataStr);
trafficLogger.logUpstreamResponse(requestId || 'unknown', '/v1/chat/completions', 200, json, 0, { rawLine: trimmed });
trafficLogger.logUpstreamResponse(
requestId || 'unknown',
'/v1/chat/completions',
200,
json,
0,
{ rawLine: trimmed },
);
// v1internal API wraps response in {"response": {"candidates": [...]}}
const responsePayload = json.response ?? json;
if (responsePayload.usageMetadata) {
@@ -1300,13 +1371,17 @@ export class ProxyService {
);
} else if (accumulatedText) {
const estimated = this.estimateTokens(accumulatedText);
this.recordUsage(accountId, effectiveTargetModel, {
promptTokenCount: 0,
candidatesTokenCount: estimated,
totalTokenCount: estimated,
}, 'openai', true).catch((err) =>
this.logger.error('Usage recording failed', err),
);
this.recordUsage(
accountId,
effectiveTargetModel,
{
promptTokenCount: 0,
candidatesTokenCount: estimated,
totalTokenCount: estimated,
},
'openai',
true,
).catch((err) => this.logger.error('Usage recording failed', err));
}
subscriber.complete();
});
@@ -1776,7 +1851,7 @@ export class ProxyService {
message: {
role: 'assistant',
// When tool_calls are present, content must be null per OpenAI spec.
content: toolCalls.length > 0 ? null : (finalContent || ''),
content: toolCalls.length > 0 ? null : finalContent || '',
tool_calls: toolCalls.length > 0 ? toolCalls : undefined,
},
finish_reason: this.mapAnthropicStopReasonToOpenAIFinishReason(
@@ -1844,7 +1919,9 @@ export class ProxyService {
if (error instanceof UpstreamRequestError) {
const status = error.status;
if (status === 401) {
this.logger.warn(`Upstream returned 401 for account ${accountId}; attempting token refresh`);
this.logger.warn(
`Upstream returned 401 for account ${accountId}; attempting token refresh`,
);
const refreshed = await this.tokenManager.forceRefreshToken(accountId);
if (refreshed) {
this.logger.log(`Token refreshed successfully for account ${accountId}; retry permitted`);
@@ -6,6 +6,7 @@ import { GoogleAPIService } from '../../../services/GoogleAPIService';
import { getServerConfig } from '../../server-config';
import { RateLimitReason, RateLimitTracker } from './rate-limit-tracker';
import { updateDynamicForwardingRules } from '../../../lib/geminiNexus/ModelMapping';
import { registerProxyAdvancedService } from '../../../ipc/proxy-advanced/service-registry';
interface TokenData {
email: string;
@@ -36,6 +37,14 @@ interface GetNextTokenOptions {
type TokenEntry = [string, TokenData];
interface CircuitBreakerEntry {
failureCount: number;
lastFailureTime: number | null;
state: 'healthy' | 'degraded' | 'unhealthy';
failureRate: number;
requestLog: Array<{ timestamp: number; success: boolean }>;
}
function normalizeProjectId(projectId: string | null | undefined): string | undefined {
if (!isString(projectId)) {
return undefined;
@@ -89,6 +98,11 @@ export class TokenManagerService implements OnModuleInit {
private accountCooldowns: Map<string, number> = new Map();
private sessionBindings: Map<string, { accountId: string; expiresAt: number }> = new Map();
private rateLimitTracker = new RateLimitTracker();
private circuitBreakerStates = new Map<string, CircuitBreakerEntry>();
private readonly circuitBreakerWindowMs = 5 * 60 * 1000;
private readonly circuitBreakerUnhealthyThreshold = 5;
private readonly circuitBreakerDegradedThreshold = 2;
private readonly circuitBreakerFailureRateThreshold = 0.5;
private shadowComparisonCount = 0;
private shadowMismatchCount = 0;
@@ -96,6 +110,10 @@ export class TokenManagerService implements OnModuleInit {
private parityErrorCount = 0;
private noGoBlocked = false;
constructor() {
registerProxyAdvancedService('tokenManager', this);
}
async onModuleInit() {
await this.loadAccounts();
}
@@ -115,6 +133,13 @@ export class TokenManagerService implements OnModuleInit {
}
}
// Prune circuit breaker states for removed accounts
for (const accountId of this.circuitBreakerStates.keys()) {
if (!this.tokens.has(accountId)) {
this.circuitBreakerStates.delete(accountId);
}
}
this.logger.log(`Token manager loaded ${count} cloud accounts into cache`);
return count;
} catch (e) {
@@ -331,18 +356,24 @@ export class TokenManagerService implements OnModuleInit {
);
}
if (candidateAccountPool.length === 0) {
// Circuit breaker: skip unhealthy accounts unless all are unhealthy
const healthyPool = candidateAccountPool.filter(
([accountId]) => this.getOrCreateCircuitBreakerEntry(accountId).state !== 'unhealthy',
);
const finalCandidatePool = healthyPool.length > 0 ? healthyPool : candidateAccountPool;
if (finalCandidatePool.length === 0) {
this.logger.warn('No eligible account found after exclusion filtering');
return null;
}
if (this.shouldExecuteShadowComparison()) {
this.executeShadowComparison(candidateAccountPool, sessionKey, model);
this.executeShadowComparison(finalCandidatePool, sessionKey, model);
}
const selectedTokenEntry = this.isParitySchedulingEnabled()
? await this.selectParityTokenCandidate(candidateAccountPool, sessionKey, model, now)
: this.selectLegacyTokenCandidate(candidateAccountPool, sessionKey, now);
? await this.selectParityTokenCandidate(finalCandidatePool, sessionKey, model, now)
: this.selectLegacyTokenCandidate(finalCandidatePool, sessionKey, now);
if (!selectedTokenEntry) {
return null;
@@ -978,6 +1009,95 @@ export class TokenManagerService implements OnModuleInit {
return this.tokens.size;
}
recordSuccess(accountId: string): void {
const resolvedId = this.resolveAccountId(accountId) ?? accountId;
const entry = this.getOrCreateCircuitBreakerEntry(resolvedId);
entry.failureCount = 0;
entry.requestLog.push({ timestamp: Date.now(), success: true });
this.computeCircuitBreakerState(entry);
}
recordFailure(accountId: string, error?: string): void {
const resolvedId = this.resolveAccountId(accountId) ?? accountId;
const entry = this.getOrCreateCircuitBreakerEntry(resolvedId);
entry.failureCount++;
entry.lastFailureTime = Date.now();
entry.requestLog.push({ timestamp: Date.now(), success: false });
this.computeCircuitBreakerState(entry);
if (entry.state === 'unhealthy') {
this.logger.warn(
`Account ${resolvedId} circuit breaker opened: failureCount=${entry.failureCount}, failureRate=${entry.failureRate.toFixed(2)}${error ? `, error=${error}` : ''}`,
);
} else if (entry.state === 'degraded') {
this.logger.warn(
`Account ${resolvedId} circuit breaker degraded: failureCount=${entry.failureCount}, failureRate=${entry.failureRate.toFixed(2)}${error ? `, error=${error}` : ''}`,
);
}
}
getCircuitBreakerStatus(): Record<string, { state: 'healthy' | 'degraded' | 'unhealthy'; failureCount: number; lastFailureTime: number | null; failureRate: number }> {
const result: Record<string, { state: 'healthy' | 'degraded' | 'unhealthy'; failureCount: number; lastFailureTime: number | null; failureRate: number }> = {};
for (const accountId of this.tokens.keys()) {
const entry = this.getOrCreateCircuitBreakerEntry(accountId);
this.computeCircuitBreakerState(entry);
result[accountId] = {
state: entry.state,
failureCount: entry.failureCount,
lastFailureTime: entry.lastFailureTime,
failureRate: entry.failureRate,
};
}
return result;
}
getParityCounters(): { totalRequests: number; matchedRequests: number; parityViolations: number; lastUpdated: string } {
return {
totalRequests: this.parityRequestCount,
matchedRequests: this.shadowComparisonCount,
parityViolations: this.parityErrorCount,
lastUpdated: new Date().toISOString(),
};
}
private getOrCreateCircuitBreakerEntry(accountId: string): CircuitBreakerEntry {
let entry = this.circuitBreakerStates.get(accountId);
if (!entry) {
entry = {
failureCount: 0,
lastFailureTime: null,
state: 'healthy',
failureRate: 0,
requestLog: [],
};
this.circuitBreakerStates.set(accountId, entry);
}
return entry;
}
private pruneCircuitBreakerLog(entry: CircuitBreakerEntry): void {
const cutoff = Date.now() - this.circuitBreakerWindowMs;
entry.requestLog = entry.requestLog.filter((r) => r.timestamp > cutoff);
}
private computeCircuitBreakerState(entry: CircuitBreakerEntry): void {
this.pruneCircuitBreakerLog(entry);
const total = entry.requestLog.length;
const failures = entry.requestLog.filter((r) => !r.success).length;
entry.failureRate = total > 0 ? failures / total : 0;
if (entry.failureCount >= this.circuitBreakerUnhealthyThreshold) {
entry.state = 'unhealthy';
} else if (
entry.failureCount >= this.circuitBreakerDegradedThreshold &&
entry.failureRate >= this.circuitBreakerFailureRateThreshold
) {
entry.state = 'degraded';
} else {
entry.state = 'healthy';
}
}
private normalizeRefreshedOauthClientKey(
currentToken: { oauth_client_key?: string; project_id?: string },
refreshedClientKey?: string,