fix: project id fallback and stream error regression (#94)
* fix(proxy): harden project id handling for upgraded accounts * fix(proxy): parse stream error payloads to preserve upstream project errors
This commit is contained in:
4 files changed
+240
-76
No files matched your search
@@ -198,7 +198,7 @@ export class GeminiClient {
|
||||
const hasNextEndpoint = index < baseUrls.length - 1;
|
||||
|
||||
if (!hasNextEndpoint || !this.shouldSwitchEndpointOnError(error)) {
|
||||
this.handleAxiosError(error, operation);
|
||||
await this.handleAxiosError(error, operation);
|
||||
}
|
||||
|
||||
this.logger.warn(
|
||||
@@ -209,33 +209,141 @@ export class GeminiClient {
|
||||
}
|
||||
}
|
||||
|
||||
this.handleAxiosError(lastError, operation);
|
||||
await this.handleAxiosError(lastError, operation);
|
||||
throw new Error(`[${operation}] unexpected control flow after upstream error handling`);
|
||||
}
|
||||
|
||||
private handleAxiosError(error: unknown, operation: string): never {
|
||||
private async handleAxiosError(error: unknown, operation: string): Promise<never> {
|
||||
if (axios.isAxiosError(error)) {
|
||||
const responseData = error.response?.data;
|
||||
const upstreamMessage = this.extractAxiosErrorMessage(responseData);
|
||||
const upstreamMessage = await this.extractAxiosErrorMessage(responseData);
|
||||
this.logger.error(
|
||||
`[${operation}] upstream request error: ${error.message} - ${this.safeStringify(responseData)}`,
|
||||
`[${operation}] upstream request error: ${error.message} - ${this.describeAxiosErrorData(
|
||||
responseData,
|
||||
)}`,
|
||||
);
|
||||
throw new Error(upstreamMessage || error.message);
|
||||
}
|
||||
this.throwAsCleanError(error);
|
||||
}
|
||||
|
||||
private extractAxiosErrorMessage(responseData: unknown): string | null {
|
||||
if (!responseData || typeof responseData !== 'object') {
|
||||
private async extractAxiosErrorMessage(responseData: unknown): Promise<string | null> {
|
||||
const fromObject = this.extractAxiosErrorMessageFromObject(responseData);
|
||||
if (fromObject) {
|
||||
return fromObject;
|
||||
}
|
||||
|
||||
if (typeof responseData === 'string') {
|
||||
return this.extractAxiosErrorMessageFromText(responseData);
|
||||
}
|
||||
|
||||
if (Buffer.isBuffer(responseData)) {
|
||||
return this.extractAxiosErrorMessageFromText(responseData.toString('utf-8'));
|
||||
}
|
||||
|
||||
if (this.isReadableStream(responseData)) {
|
||||
const streamText = await this.readStreamAsText(responseData);
|
||||
return streamText ? this.extractAxiosErrorMessageFromText(streamText) : null;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private extractAxiosErrorMessageFromObject(responseData: unknown): string | null {
|
||||
if (!responseData || typeof responseData !== 'object' || this.isReadableStream(responseData)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const errorRecord = (responseData as { error?: unknown }).error;
|
||||
if (!errorRecord || typeof errorRecord !== 'object') {
|
||||
if (errorRecord && typeof errorRecord === 'object') {
|
||||
const message = (errorRecord as { message?: unknown }).message;
|
||||
if (typeof message === 'string' && message.trim() !== '') {
|
||||
return message.trim();
|
||||
}
|
||||
}
|
||||
|
||||
const message = (responseData as { message?: unknown }).message;
|
||||
if (typeof message === 'string' && message.trim() !== '') {
|
||||
return message.trim();
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private extractAxiosErrorMessageFromText(rawText: string): string | null {
|
||||
const text = rawText.trim();
|
||||
if (!text) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const message = (errorRecord as { message?: unknown }).message;
|
||||
return typeof message === 'string' && message.trim() !== '' ? message : null;
|
||||
const sseLines = text
|
||||
.split('\n')
|
||||
.map((line) => line.trim())
|
||||
.filter((line) => line.startsWith('data:'));
|
||||
|
||||
for (const line of sseLines) {
|
||||
const payload = line.slice(5).trim();
|
||||
if (!payload || payload === '[DONE]') {
|
||||
continue;
|
||||
}
|
||||
const parsed = this.tryParseJson(payload);
|
||||
const message = this.extractAxiosErrorMessageFromObject(parsed);
|
||||
if (message) {
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
||||
const parsed = this.tryParseJson(text);
|
||||
const fromJson = this.extractAxiosErrorMessageFromObject(parsed);
|
||||
if (fromJson) {
|
||||
return fromJson;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private tryParseJson(value: string): unknown {
|
||||
try {
|
||||
return JSON.parse(value);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private isReadableStream(value: unknown): value is NodeJS.ReadableStream {
|
||||
return (
|
||||
Boolean(value) &&
|
||||
typeof value === 'object' &&
|
||||
typeof (value as { pipe?: unknown }).pipe === 'function'
|
||||
);
|
||||
}
|
||||
|
||||
private async readStreamAsText(stream: NodeJS.ReadableStream): Promise<string | null> {
|
||||
return new Promise((resolve) => {
|
||||
let buffer = '';
|
||||
const maxChars = 512 * 1024;
|
||||
|
||||
stream.on('data', (chunk: Buffer | string) => {
|
||||
const text = Buffer.isBuffer(chunk) ? chunk.toString('utf-8') : String(chunk);
|
||||
if (buffer.length >= maxChars) {
|
||||
return;
|
||||
}
|
||||
buffer += text;
|
||||
if (buffer.length > maxChars) {
|
||||
buffer = buffer.slice(0, maxChars);
|
||||
}
|
||||
});
|
||||
|
||||
stream.on('end', () => resolve(buffer));
|
||||
stream.on('error', () => resolve(null));
|
||||
});
|
||||
}
|
||||
|
||||
private describeAxiosErrorData(responseData: unknown): string {
|
||||
if (this.isReadableStream(responseData)) {
|
||||
return '[stream]';
|
||||
}
|
||||
return this.safeStringify(responseData);
|
||||
}
|
||||
|
||||
private resolveAxiosProxy(upstreamProxyUrl?: string): AxiosProxyConfig | false | undefined {
|
||||
|
||||
@@ -26,12 +26,19 @@ function normalizeProjectId(projectId: string | null | undefined): string | unde
|
||||
return undefined;
|
||||
}
|
||||
|
||||
// Legacy upgrades may carry malformed resource names (e.g. "projects/").
|
||||
// The internal gateway expects a plain project id, not a resource path.
|
||||
if (/^projects(?:\/.*)?$/i.test(trimmedProjectId)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return trimmedProjectId;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class TokenManagerService implements OnModuleInit {
|
||||
private readonly logger = new Logger(TokenManagerService.name);
|
||||
private readonly defaultFallbackProjectId = 'silver-orbit-5m7qc';
|
||||
private currentIndex = 0;
|
||||
private readonly stickySessionTtlMs = 10 * 60 * 1000;
|
||||
private readonly rateLimitCooldownMs = 5 * 60 * 1000;
|
||||
@@ -178,6 +185,8 @@ export class TokenManagerService implements OnModuleInit {
|
||||
sessionKey?: string,
|
||||
): Promise<CloudAccount | null> {
|
||||
try {
|
||||
let effectiveProjectId: string | undefined;
|
||||
|
||||
// Check if token needs refresh (expires in < 5 minutes)
|
||||
if (nowSeconds >= tokenData.expiry_timestamp - 300) {
|
||||
this.logger.log(`Token for ${tokenData.email} is close to expiry; refreshing...`);
|
||||
@@ -199,19 +208,19 @@ export class TokenManagerService implements OnModuleInit {
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
normalizeProjectId(tokenData.project_id) === undefined
|
||||
) {
|
||||
if (normalizeProjectId(tokenData.project_id) === undefined) {
|
||||
tokenData.project_id = undefined;
|
||||
}
|
||||
effectiveProjectId = tokenData.project_id;
|
||||
|
||||
if (!tokenData.project_id) {
|
||||
if (!effectiveProjectId) {
|
||||
try {
|
||||
const fetchedProjectId = await GoogleAPIService.fetchProjectId(tokenData.access_token);
|
||||
const normalizedProjectId = normalizeProjectId(fetchedProjectId);
|
||||
|
||||
if (normalizedProjectId) {
|
||||
tokenData.project_id = normalizedProjectId;
|
||||
effectiveProjectId = normalizedProjectId;
|
||||
await this.saveTokenState(accountId, tokenData);
|
||||
this.tokens.set(accountId, tokenData);
|
||||
this.logger.log(`Resolved project ID for ${tokenData.email}: ${normalizedProjectId}`);
|
||||
@@ -225,6 +234,14 @@ export class TokenManagerService implements OnModuleInit {
|
||||
}
|
||||
}
|
||||
|
||||
if (!effectiveProjectId) {
|
||||
const fallbackProjectId = this.getFallbackProjectId();
|
||||
effectiveProjectId = fallbackProjectId;
|
||||
this.logger.warn(
|
||||
`Using fallback project ID for ${tokenData.email} without persistence: ${fallbackProjectId}`,
|
||||
);
|
||||
}
|
||||
|
||||
this.logger.log(`Using account: ${tokenData.email}`);
|
||||
if (sessionKey) {
|
||||
this.sessionBindings.set(sessionKey, {
|
||||
@@ -245,7 +262,7 @@ export class TokenManagerService implements OnModuleInit {
|
||||
token_type: tokenData.token_type,
|
||||
expires_in: tokenData.expires_in,
|
||||
expiry_timestamp: tokenData.expiry_timestamp,
|
||||
project_id: tokenData.project_id,
|
||||
project_id: effectiveProjectId,
|
||||
session_id: tokenData.session_id,
|
||||
upstream_proxy_url: tokenData.upstream_proxy_url,
|
||||
},
|
||||
@@ -329,4 +346,13 @@ export class TokenManagerService implements OnModuleInit {
|
||||
getAccountCount(): number {
|
||||
return this.tokens.size;
|
||||
}
|
||||
|
||||
private getFallbackProjectId(): string {
|
||||
const fromEnv = process.env.PROXY_FALLBACK_PROJECT_ID?.trim();
|
||||
const normalizedFromEnv = normalizeProjectId(fromEnv);
|
||||
if (normalizedFromEnv) {
|
||||
return normalizedFromEnv;
|
||||
}
|
||||
return this.defaultFallbackProjectId;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user