refactor: 移除 CredentialCacheManager 及相关缓存逻辑
重构多个 provider 模块，移除了 CredentialCacheManager 的依赖及相关内存缓存逻辑，改为直接读写文件系统。简化了凭证管理流程，减少了代码复杂度。 - 移除 provider-pool-manager 中的快速预检逻辑 - 移除 service-manager 中的凭证缓存初始化及同步逻辑 - 修改各 provider 核心模块的凭证加载和保存逻辑，直接操作文件系统 - 移除并发刷新时的去重锁和缓存同步机制
This commit is contained in:
hex2077
parent
6ba6f42a94
commit
56ef11a168
8 changed files with 140 additions and 1470 deletions
|
|
@ -11,7 +11,6 @@ import { countTokens } from '@anthropic-ai/tokenizer';
|
|||
import { configureAxiosProxy } from '../../utils/proxy-utils.js';
|
||||
import { isRetryableNetworkError, MODEL_PROVIDER } from '../../utils/common.js';
|
||||
import { getProviderPoolManager } from '../../services/service-manager.js';
|
||||
import { CredentialCacheManager } from '../../utils/credential-cache-manager.js';
|
||||
|
||||
const KIRO_THINKING = {
|
||||
MAX_BUDGET_TOKENS: 24576,
|
||||
|
|
@ -465,14 +464,10 @@ async initializeAuth(forceRefresh = false) {
|
|||
return;
|
||||
}
|
||||
|
||||
// 获取凭证文件路径，用于去重锁的 key
|
||||
// 获取凭证文件路径
|
||||
const tokenFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AUTH_TOKEN_FILE);
|
||||
|
||||
// 获取凭证缓存管理器
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'claude-kiro-oauth';
|
||||
|
||||
// Helper to load credentials from a file (fallback when cache miss)
|
||||
// Helper to load credentials from a file
|
||||
const loadCredentialsFromFile = async (filePath) => {
|
||||
try {
|
||||
const fileContent = await fs.readFile(filePath, 'utf8');
|
||||
|
|
@ -507,55 +502,41 @@ async initializeAuth(forceRefresh = false) {
|
|||
}
|
||||
};
|
||||
|
||||
// Helper to save credentials - 使用内存缓存替代直接文件写入
|
||||
// Helper to save credentials
|
||||
const saveCredentialsToFile = async (filePath, newData) => {
|
||||
// 优先更新内存缓存
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
const entry = credentialCache.getCredentials(providerType, this.uuid);
|
||||
if (entry) {
|
||||
const mergedData = { ...entry.credentials, ...newData };
|
||||
credentialCache.updateCredentials(providerType, this.uuid, mergedData, filePath);
|
||||
console.info(`[Kiro Auth] Updated credentials in memory cache: ${this.uuid}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// 如果没有缓存条目，使用内存锁方式写入
|
||||
await credentialCache.withMemoryLock(`kiro-save:${filePath}`, async () => {
|
||||
let existingData = {};
|
||||
let existingData = {};
|
||||
try {
|
||||
const fileContent = await fs.readFile(filePath, 'utf8');
|
||||
try {
|
||||
const fileContent = await fs.readFile(filePath, 'utf8');
|
||||
existingData = JSON.parse(fileContent);
|
||||
} catch (parseError) {
|
||||
console.warn('[Kiro Auth] JSON parse failed, attempting repair...');
|
||||
try {
|
||||
existingData = JSON.parse(fileContent);
|
||||
} catch (parseError) {
|
||||
console.warn('[Kiro Auth] JSON parse failed, attempting repair...');
|
||||
try {
|
||||
const repaired = repairJson(fileContent);
|
||||
existingData = JSON.parse(repaired);
|
||||
console.info('[Kiro Auth] JSON repair successful');
|
||||
} catch (repairError) {
|
||||
console.warn('[Kiro Auth] JSON repair failed, attempting field extraction...');
|
||||
const extracted = extractCredentialsFromCorruptedJson(fileContent);
|
||||
if (extracted) {
|
||||
existingData = extracted;
|
||||
console.info('[Kiro Auth] Field extraction successful');
|
||||
} else {
|
||||
console.error('[Kiro Auth] All recovery methods failed:', repairError.message);
|
||||
existingData = {};
|
||||
}
|
||||
const repaired = repairJson(fileContent);
|
||||
existingData = JSON.parse(repaired);
|
||||
console.info('[Kiro Auth] JSON repair successful');
|
||||
} catch (repairError) {
|
||||
console.warn('[Kiro Auth] JSON repair failed, attempting field extraction...');
|
||||
const extracted = extractCredentialsFromCorruptedJson(fileContent);
|
||||
if (extracted) {
|
||||
existingData = extracted;
|
||||
console.info('[Kiro Auth] Field extraction successful');
|
||||
} else {
|
||||
console.error('[Kiro Auth] All recovery methods failed:', repairError.message);
|
||||
existingData = {};
|
||||
}
|
||||
}
|
||||
} catch (readError) {
|
||||
if (readError.code === 'ENOENT') {
|
||||
console.debug(`[Kiro Auth] Token file not found, creating new one: ${filePath}`);
|
||||
} else {
|
||||
console.warn(`[Kiro Auth] Could not read existing token file ${filePath}: ${readError.message}`);
|
||||
}
|
||||
}
|
||||
const mergedData = { ...existingData, ...newData };
|
||||
await fs.writeFile(filePath, JSON.stringify(mergedData, null, 2), 'utf8');
|
||||
console.info(`[Kiro Auth] Updated token file: ${filePath}`);
|
||||
});
|
||||
} catch (readError) {
|
||||
if (readError.code === 'ENOENT') {
|
||||
console.debug(`[Kiro Auth] Token file not found, creating new one: ${filePath}`);
|
||||
} else {
|
||||
console.warn(`[Kiro Auth] Could not read existing token file ${filePath}: ${readError.message}`);
|
||||
}
|
||||
}
|
||||
const mergedData = { ...existingData, ...newData };
|
||||
await fs.writeFile(filePath, JSON.stringify(mergedData, null, 2), 'utf8');
|
||||
console.info(`[Kiro Auth] Updated token file: ${filePath}`);
|
||||
};
|
||||
|
||||
try {
|
||||
|
|
@ -568,43 +549,34 @@ async initializeAuth(forceRefresh = false) {
|
|||
this.base64Creds = null;
|
||||
}
|
||||
|
||||
// Priority 2: 尝试从内存缓存加载凭证
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, this.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
Object.assign(mergedCredentials, cachedEntry.credentials);
|
||||
console.info(`[Kiro Auth] Successfully loaded credentials from memory cache: ${this.uuid}`);
|
||||
// 从文件加载
|
||||
const targetFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AUTH_TOKEN_FILE);
|
||||
const dirPath = path.dirname(targetFilePath);
|
||||
const targetFileName = path.basename(targetFilePath);
|
||||
|
||||
console.debug(`[Kiro Auth] Loading credentials from directory: ${dirPath}`);
|
||||
|
||||
try {
|
||||
const targetCredentials = await loadCredentialsFromFile(targetFilePath);
|
||||
if (targetCredentials) {
|
||||
Object.assign(mergedCredentials, targetCredentials);
|
||||
console.info(`[Kiro Auth] Successfully loaded OAuth credentials from ${targetFilePath}`);
|
||||
}
|
||||
} else {
|
||||
// Priority 3: 从文件加载（缓存未命中时的回退）
|
||||
const targetFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AUTH_TOKEN_FILE);
|
||||
const dirPath = path.dirname(targetFilePath);
|
||||
const targetFileName = path.basename(targetFilePath);
|
||||
|
||||
console.debug(`[Kiro Auth] Cache miss, loading credentials from directory: ${dirPath}`);
|
||||
|
||||
try {
|
||||
const targetCredentials = await loadCredentialsFromFile(targetFilePath);
|
||||
if (targetCredentials) {
|
||||
Object.assign(mergedCredentials, targetCredentials);
|
||||
console.info(`[Kiro Auth] Successfully loaded OAuth credentials from ${targetFilePath}`);
|
||||
}
|
||||
|
||||
const files = await fs.readdir(dirPath);
|
||||
for (const file of files) {
|
||||
if (file.endsWith('.json') && file !== targetFileName) {
|
||||
const filePath = path.join(dirPath, file);
|
||||
const credentials = await loadCredentialsFromFile(filePath);
|
||||
if (credentials) {
|
||||
credentials.expiresAt = mergedCredentials.expiresAt;
|
||||
Object.assign(mergedCredentials, credentials);
|
||||
console.debug(`[Kiro Auth] Loaded Client credentials from ${file}`);
|
||||
}
|
||||
const files = await fs.readdir(dirPath);
|
||||
for (const file of files) {
|
||||
if (file.endsWith('.json') && file !== targetFileName) {
|
||||
const filePath = path.join(dirPath, file);
|
||||
const credentials = await loadCredentialsFromFile(filePath);
|
||||
if (credentials) {
|
||||
credentials.expiresAt = mergedCredentials.expiresAt;
|
||||
Object.assign(mergedCredentials, credentials);
|
||||
console.debug(`[Kiro Auth] Loaded Client credentials from ${file}`);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn(`[Kiro Auth] Error loading credentials from directory ${dirPath}: ${error.message}`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn(`[Kiro Auth] Error loading credentials from directory ${dirPath}: ${error.message}`);
|
||||
}
|
||||
|
||||
// Apply loaded credentials
|
||||
|
|
@ -636,15 +608,7 @@ async initializeAuth(forceRefresh = false) {
|
|||
throw new Error('No refresh token available to refresh access token.');
|
||||
}
|
||||
|
||||
// 使用内存锁替代文件锁进行去重
|
||||
const dedupeKey = `kiro-token-refresh:${tokenFilePath}`;
|
||||
await credentialCache.withDeduplication(dedupeKey, async () => {
|
||||
await this._doTokenRefresh(saveCredentialsToFile, tokenFilePath);
|
||||
});
|
||||
|
||||
// 刷新完成后（无论是自己执行还是等待其他请求），都从缓存重新加载凭证
|
||||
// 确保所有并发请求都能获取到最新的 token
|
||||
await this._reloadCredentialsAfterRefresh(tokenFilePath);
|
||||
await this._doTokenRefresh(saveCredentialsToFile, tokenFilePath);
|
||||
}
|
||||
|
||||
if (!this.accessToken) {
|
||||
|
|
@ -709,58 +673,6 @@ async initializeAuth(forceRefresh = false) {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 在并发刷新完成后重新加载凭证（内部方法）
|
||||
* @param {string} tokenFilePath - 凭证文件路径
|
||||
*/
|
||||
async _reloadCredentialsAfterRefresh(tokenFilePath) {
|
||||
// 优先从内存缓存加载
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'claude-kiro-oauth';
|
||||
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, this.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
this.accessToken = cachedEntry.credentials.accessToken;
|
||||
this.refreshToken = cachedEntry.credentials.refreshToken;
|
||||
this.expiresAt = cachedEntry.credentials.expiresAt;
|
||||
if (cachedEntry.credentials.profileArn) {
|
||||
this.profileArn = cachedEntry.credentials.profileArn;
|
||||
}
|
||||
console.debug('[Kiro Auth] Credentials reloaded from memory cache after concurrent refresh');
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// 回退到文件加载
|
||||
try {
|
||||
const fileContent = await fs.readFile(tokenFilePath, 'utf8');
|
||||
let credentials;
|
||||
try {
|
||||
credentials = JSON.parse(fileContent);
|
||||
} catch (parseError) {
|
||||
console.warn('[Kiro Auth] JSON parse failed, attempting repair...');
|
||||
try {
|
||||
const repaired = repairJson(fileContent);
|
||||
credentials = JSON.parse(repaired);
|
||||
console.info('[Kiro Auth] JSON repair successful');
|
||||
} catch (repairError) {
|
||||
console.error('[Kiro Auth] JSON repair failed:', repairError.message);
|
||||
throw new Error(`Failed to parse credentials file after repair attempt: ${repairError.message}`);
|
||||
}
|
||||
}
|
||||
this.accessToken = credentials.accessToken;
|
||||
this.refreshToken = credentials.refreshToken;
|
||||
this.expiresAt = credentials.expiresAt;
|
||||
if (credentials.profileArn) {
|
||||
this.profileArn = credentials.profileArn;
|
||||
}
|
||||
console.debug('[Kiro Auth] Credentials reloaded from file after concurrent refresh');
|
||||
} catch (error) {
|
||||
console.warn(`[Kiro Auth] Failed to reload credentials after refresh: ${error.message}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract text content from OpenAI message format
|
||||
|
|
@ -2657,22 +2569,12 @@ async initializeAuth(forceRefresh = false) {
|
|||
* 后台异步刷新 token（不阻塞当前请求）
|
||||
*/
|
||||
triggerBackgroundRefresh() {
|
||||
const tokenFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AH_TOKEN_FILE);
|
||||
const dedupeKey = `kiro-token-refresh:${tokenFilePath}`;
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
|
||||
// 使用 withDeduplication 确保只有一个刷新任务在执行
|
||||
credentialCache.withDeduplication(dedupeKey, async () => {
|
||||
console.log('[Kiro] Background token refresh started...');
|
||||
try {
|
||||
await this.initializeAuth(true);
|
||||
console.log('[Kiro] Background token refresh completed successfully');
|
||||
} catch (error) {
|
||||
console.error('[Kiro] Background token refresh failed:', error.message);
|
||||
// 后台刷新失败不抛出错误，下次请求会重试
|
||||
}
|
||||
}).catch(() => {
|
||||
// 忽略后台刷新错误，不影响当前请求
|
||||
console.log('[Kiro] Background token refresh started...');
|
||||
this.initializeAuth(true).then(() => {
|
||||
console.log('[Kiro] Background token refresh completed successfully');
|
||||
}).catch((error) => {
|
||||
console.error('[Kiro] Background token refresh failed:', error.message);
|
||||
// 后台刷新失败不抛出错误，下次请求会重试
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ import { getProviderModels } from '../provider-models.js';
|
|||
import { handleGeminiAntigravityOAuth } from '../../auth/oauth-handlers.js';
|
||||
import { getProxyConfigForProvider, getGoogleAuthProxyConfig } from '../../utils/proxy-utils.js';
|
||||
import { cleanJsonSchemaProperties } from '../../converters/utils.js';
|
||||
import { CredentialCacheManager } from '../../utils/credential-cache-manager.js';
|
||||
|
||||
// 配置 HTTP/HTTPS agent 限制连接池大小，避免资源泄漏
|
||||
const httpAgent = new http.Agent({
|
||||
|
|
@ -777,9 +776,6 @@ export class AntigravityApiService {
|
|||
}
|
||||
|
||||
async initializeAuth(forceRefresh = false) {
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'gemini-antigravity';
|
||||
|
||||
// 检查是否需要刷新 Token
|
||||
const needsRefresh = forceRefresh || this.isTokenExpiringSoon();
|
||||
|
||||
|
|
@ -788,64 +784,19 @@ export class AntigravityApiService {
|
|||
return;
|
||||
}
|
||||
|
||||
// Antigravity 不支持 base64 配置，直接使用文件路径
|
||||
|
||||
const credPath = this.oauthCredsFilePath || path.join(os.homedir(), CREDENTIALS_DIR, CREDENTIALS_FILE);
|
||||
try {
|
||||
// 优先从内存缓存加载
|
||||
let credentials = null;
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, this.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
credentials = cachedEntry.credentials;
|
||||
console.log('[Antigravity Auth] Loaded credentials from memory cache');
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: 从文件加载
|
||||
if (!credentials) {
|
||||
const data = await fs.readFile(credPath, "utf8");
|
||||
credentials = JSON.parse(data);
|
||||
console.log('[Antigravity Auth] Loaded credentials from file');
|
||||
}
|
||||
|
||||
const data = await fs.readFile(credPath, "utf8");
|
||||
const credentials = JSON.parse(data);
|
||||
this.authClient.setCredentials(credentials);
|
||||
console.log('[Antigravity Auth] Authentication configured successfully from file.');
|
||||
|
||||
if (needsRefresh) {
|
||||
// 使用去重锁：多个并发刷新请求只执行一次，共享结果
|
||||
const dedupeKey = `antigravity-token-refresh:${credPath}`;
|
||||
await credentialCache.withDeduplication(dedupeKey, async () => {
|
||||
console.log('[Antigravity Auth] Token expiring soon or force refresh requested. Refreshing token...');
|
||||
const { credentials: newCredentials } = await this.authClient.refreshAccessToken();
|
||||
this.authClient.setCredentials(newCredentials);
|
||||
// 保存刷新后的凭证（优先保存到内存缓存）
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
credentialCache.updateCredentials(providerType, this.uuid, newCredentials, credPath);
|
||||
console.log(`[Antigravity Auth] Token refreshed and saved to memory cache: ${this.uuid}`);
|
||||
} else {
|
||||
await this._saveCredentialsToFile(credPath, newCredentials);
|
||||
}
|
||||
console.log(`[Antigravity Auth] Token refreshed and saved to ${credPath} successfully.`);
|
||||
});
|
||||
|
||||
// 如果是等待其他请求完成的刷新，需要重新加载凭证
|
||||
// 因为 withDeduplication 只让第一个调用者执行刷新并更新自己的内存状态
|
||||
// 其他等待者需要从缓存重新加载
|
||||
if (this.isTokenExpiringSoon()) {
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, this.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
this.authClient.setCredentials(cachedEntry.credentials);
|
||||
console.log('[Antigravity Auth] Credentials reloaded from memory cache after concurrent refresh');
|
||||
}
|
||||
} else {
|
||||
const refreshedData = await fs.readFile(credPath, "utf8");
|
||||
const refreshedCredentials = JSON.parse(refreshedData);
|
||||
this.authClient.setCredentials(refreshedCredentials);
|
||||
console.log('[Antigravity Auth] Credentials reloaded from file after concurrent refresh');
|
||||
}
|
||||
}
|
||||
console.log('[Antigravity Auth] Token expiring soon or force refresh requested. Refreshing token...');
|
||||
const { credentials: newCredentials } = await this.authClient.refreshAccessToken();
|
||||
this.authClient.setCredentials(newCredentials);
|
||||
await this._saveCredentialsToFile(credPath, newCredentials);
|
||||
console.log(`[Antigravity Auth] Token refreshed and saved to ${credPath} successfully.`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[Antigravity Auth] Error initializing authentication:', error.code);
|
||||
|
|
@ -921,30 +872,18 @@ export class AntigravityApiService {
|
|||
}
|
||||
|
||||
/**
|
||||
* 保存凭证到文件（优先使用内存缓存）
|
||||
* 保存凭证到文件
|
||||
* @param {string} filePath - 凭证文件路径
|
||||
* @param {Object} credentials - 凭证数据
|
||||
*/
|
||||
async _saveCredentialsToFile(filePath, credentials) {
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'gemini-antigravity';
|
||||
|
||||
// 优先保存到内存缓存
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
credentialCache.updateCredentials(providerType, this.uuid, credentials, filePath);
|
||||
console.log(`[Antigravity Auth] Credentials saved to memory cache: ${this.uuid}`);
|
||||
return;
|
||||
try {
|
||||
await fs.writeFile(filePath, JSON.stringify(credentials, null, 2));
|
||||
console.log(`[Antigravity Auth] Credentials saved to ${filePath}`);
|
||||
} catch (error) {
|
||||
console.error(`[Antigravity Auth] Failed to save credentials to ${filePath}: ${error.message}`);
|
||||
throw error;
|
||||
}
|
||||
|
||||
// Fallback: 使用内存锁的文件写入
|
||||
await credentialCache.withMemoryLock(`antigravity-save:${filePath}`, async () => {
|
||||
try {
|
||||
await fs.writeFile(filePath, JSON.stringify(credentials, null, 2));
|
||||
console.log(`[Antigravity Auth] Credentials saved to ${filePath}`);
|
||||
} catch (error) {
|
||||
console.error(`[Antigravity Auth] Failed to save credentials to ${filePath}: ${error.message}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async discoverProjectAndModels() {
|
||||
|
|
|
|||
|
|
@ -10,7 +10,6 @@ import { API_ACTIONS, formatExpiryTime, isRetryableNetworkError } from '../../ut
|
|||
import { getProviderModels } from '../provider-models.js';
|
||||
import { handleGeminiCliOAuth } from '../../auth/oauth-handlers.js';
|
||||
import { getProxyConfigForProvider, getGoogleAuthProxyConfig } from '../../utils/proxy-utils.js';
|
||||
import { CredentialCacheManager } from '../../utils/credential-cache-manager.js';
|
||||
|
||||
// 配置 HTTP/HTTPS agent 限制连接池大小，避免资源泄漏
|
||||
const httpAgent = new http.Agent({
|
||||
|
|
@ -273,25 +272,11 @@ export class GeminiApiService {
|
|||
|
||||
if (forceRefresh) {
|
||||
console.log('[Gemini Auth] Forcing token refresh...');
|
||||
|
||||
// 使用去重锁：多个并发刷新请求只执行一次，共享结果
|
||||
const dedupeKey = `gemini-token-refresh:${credPath}`;
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
await credentialCache.withDeduplication(dedupeKey, async () => {
|
||||
const { credentials: newCredentials } = await this.authClient.refreshAccessToken();
|
||||
this.authClient.setCredentials(newCredentials);
|
||||
// Save refreshed credentials back to file (with file locking)
|
||||
await this._saveCredentialsToFile(credPath, newCredentials);
|
||||
console.log('[Gemini Auth] Token refreshed and saved successfully.');
|
||||
});
|
||||
|
||||
// 如果是等待其他请求完成的刷新，需要重新加载凭证
|
||||
if (this.isExpiryDateNear()) {
|
||||
const refreshedData = await fs.readFile(credPath, "utf8");
|
||||
const refreshedCredentials = JSON.parse(refreshedData);
|
||||
this.authClient.setCredentials(refreshedCredentials);
|
||||
console.log('[Gemini Auth] Credentials reloaded after concurrent refresh');
|
||||
}
|
||||
const { credentials: newCredentials } = await this.authClient.refreshAccessToken();
|
||||
this.authClient.setCredentials(newCredentials);
|
||||
// Save refreshed credentials back to file
|
||||
await this._saveCredentialsToFile(credPath, newCredentials);
|
||||
console.log('[Gemini Auth] Token refreshed and saved successfully.');
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[Gemini Auth] Error initializing authentication:', error.code);
|
||||
|
|
@ -637,30 +622,18 @@ export class GeminiApiService {
|
|||
}
|
||||
|
||||
/**
|
||||
* 保存凭证到文件（使用内存缓存优先）
|
||||
* 保存凭证到文件
|
||||
* @param {string} filePath - 凭证文件路径
|
||||
* @param {Object} credentials - 凭证数据
|
||||
*/
|
||||
async _saveCredentialsToFile(filePath, credentials) {
|
||||
// 优先更新内存缓存
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'gemini-cli-oauth';
|
||||
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
credentialCache.updateCredentials(providerType, this.uuid, credentials, filePath);
|
||||
console.log(`[Gemini Auth] Credentials saved to memory cache: ${this.uuid}`);
|
||||
return;
|
||||
try {
|
||||
await fs.writeFile(filePath, JSON.stringify(credentials, null, 2));
|
||||
console.log(`[Gemini Auth] Credentials saved to ${filePath}`);
|
||||
} catch (error) {
|
||||
console.error(`[Gemini Auth] Failed to save credentials to ${filePath}: ${error.message}`);
|
||||
throw error;
|
||||
}
|
||||
|
||||
// 回退到使用内存锁写入文件
|
||||
await credentialCache.withMemoryLock(`gemini-save:${filePath}`, async () => {
|
||||
try {
|
||||
await fs.writeFile(filePath, JSON.stringify(credentials, null, 2));
|
||||
console.log(`[Gemini Auth] Credentials saved to ${filePath}`);
|
||||
} catch (error) {
|
||||
console.error(`[Gemini Auth] Failed to save credentials to ${filePath}: ${error.message}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ import * as path from 'path';
|
|||
import * as os from 'os';
|
||||
import { configureAxiosProxy } from '../../utils/proxy-utils.js';
|
||||
import { isRetryableNetworkError } from '../../utils/common.js';
|
||||
import { CredentialCacheManager } from '../../utils/credential-cache-manager.js';
|
||||
|
||||
// iFlow API 端点
|
||||
const IFLOW_API_BASE_URL = 'https://apis.iflow.cn/v1';
|
||||
|
|
@ -137,51 +136,28 @@ async function saveTokenToFile(filePath, tokenStorage, uuid = null) {
|
|||
? filePath
|
||||
: path.join(process.cwd(), filePath);
|
||||
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'openai-iflow';
|
||||
try {
|
||||
// 确保目录存在
|
||||
const dir = path.dirname(absolutePath);
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
|
||||
// 优先保存到内存缓存
|
||||
if (uuid && credentialCache.hasCredentials(providerType, uuid)) {
|
||||
// 写入文件
|
||||
const json = tokenStorage.toJSON();
|
||||
|
||||
// 验证关键字段是否存在
|
||||
if (!json.refresh_token || json.refresh_token.trim() === '') {
|
||||
console.error('[iFlow] WARNING: Attempting to save token with empty refresh_token!');
|
||||
console.error('[iFlow] WARNING: Attempting to save token file with empty refresh_token!');
|
||||
}
|
||||
if (!json.apiKey || json.apiKey.trim() === '') {
|
||||
console.error('[iFlow] WARNING: Attempting to save token with empty apiKey!');
|
||||
console.error('[iFlow] WARNING: Attempting to save token file with empty apiKey!');
|
||||
}
|
||||
|
||||
credentialCache.updateCredentials(providerType, uuid, json, absolutePath);
|
||||
console.log(`[iFlow] Token saved to memory cache: ${uuid} (refresh_token: ${json.refresh_token ? json.refresh_token.substring(0, 8) + '...' : 'EMPTY'})`);
|
||||
return;
|
||||
await fs.writeFile(absolutePath, JSON.stringify(json, null, 2), 'utf-8');
|
||||
|
||||
console.log(`[iFlow] Token saved to: ${filePath} (refresh_token: ${json.refresh_token ? json.refresh_token.substring(0, 8) + '...' : 'EMPTY'})`);
|
||||
} catch (error) {
|
||||
throw new Error(`[iFlow] Failed to save token to file: ${error.message}`);
|
||||
}
|
||||
|
||||
// Fallback: 使用内存锁的文件写入
|
||||
await credentialCache.withMemoryLock(`iflow-save:${absolutePath}`, async () => {
|
||||
try {
|
||||
// 确保目录存在
|
||||
const dir = path.dirname(absolutePath);
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
|
||||
// 写入文件
|
||||
const json = tokenStorage.toJSON();
|
||||
|
||||
// 验证关键字段是否存在
|
||||
if (!json.refresh_token || json.refresh_token.trim() === '') {
|
||||
console.error('[iFlow] WARNING: Attempting to save token file with empty refresh_token!');
|
||||
}
|
||||
if (!json.apiKey || json.apiKey.trim() === '') {
|
||||
console.error('[iFlow] WARNING: Attempting to save token file with empty apiKey!');
|
||||
}
|
||||
|
||||
await fs.writeFile(absolutePath, JSON.stringify(json, null, 2), 'utf-8');
|
||||
|
||||
console.log(`[iFlow] Token saved to: ${filePath} (refresh_token: ${json.refresh_token ? json.refresh_token.substring(0, 8) + '...' : 'EMPTY'})`);
|
||||
} catch (error) {
|
||||
throw new Error(`[iFlow] Failed to save token to file: ${error.message}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// ==================== Token 刷新逻辑 ====================
|
||||
|
|
@ -528,9 +504,6 @@ export class IFlowApiService {
|
|||
* @param {boolean} forceRefresh - 是否强制刷新 Token
|
||||
*/
|
||||
async initializeAuth(forceRefresh = false) {
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'openai-iflow';
|
||||
|
||||
// 如果已有 API Key 且不强制刷新，直接返回
|
||||
if (this.apiKey && !forceRefresh) return;
|
||||
|
||||
|
|
@ -540,18 +513,7 @@ export class IFlowApiService {
|
|||
}
|
||||
|
||||
try {
|
||||
// 优先从内存缓存加载
|
||||
let tokenData = null;
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, this.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
tokenData = cachedEntry.credentials;
|
||||
this.tokenStorage = IFlowTokenStorage.fromJSON(tokenData);
|
||||
console.log('[iFlow Auth] Loaded credentials from memory cache');
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: 从文件加载
|
||||
// 从文件加载
|
||||
if (!this.tokenStorage) {
|
||||
this.tokenStorage = await loadTokenFromFile(this.tokenFilePath);
|
||||
console.log('[iFlow Auth] Loaded credentials from file');
|
||||
|
|
@ -589,9 +551,6 @@ export class IFlowApiService {
|
|||
* @returns {Promise<boolean>} - 是否执行了刷新
|
||||
*/
|
||||
async _checkAndRefreshTokenIfNeeded() {
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'openai-iflow';
|
||||
|
||||
if (!this.tokenStorage) {
|
||||
return false;
|
||||
}
|
||||
|
|
@ -611,33 +570,7 @@ export class IFlowApiService {
|
|||
console.log('[iFlow] Token is expiring soon, attempting refresh...');
|
||||
|
||||
try {
|
||||
// 使用去重锁：多个并发刷新请求只执行一次,共享结果
|
||||
const dedupeKey = `iflow-token-refresh:${this.tokenFilePath}`;
|
||||
await credentialCache.withDeduplication(dedupeKey, async () => {
|
||||
await this._refreshOAuthTokens();
|
||||
});
|
||||
|
||||
// 如果是等待其他请求完成的刷新，需要重新加载凭证
|
||||
if (this.isExpiryDateNear()) {
|
||||
if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, this.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
this.tokenStorage = IFlowTokenStorage.fromJSON(cachedEntry.credentials);
|
||||
this.apiKey = this.tokenStorage.apiKey;
|
||||
this.axiosInstance.defaults.headers['Authorization'] = `Bearer ${this.apiKey}`;
|
||||
console.log('[iFlow] Credentials reloaded from memory cache after concurrent refresh');
|
||||
}
|
||||
} else {
|
||||
const refreshedStorage = await loadTokenFromFile(this.tokenFilePath);
|
||||
if (refreshedStorage && refreshedStorage.apiKey) {
|
||||
this.tokenStorage = refreshedStorage;
|
||||
this.apiKey = refreshedStorage.apiKey;
|
||||
this.axiosInstance.defaults.headers['Authorization'] = `Bearer ${this.apiKey}`;
|
||||
console.log('[iFlow] Credentials reloaded from file after concurrent refresh');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await this._refreshOAuthTokens();
|
||||
return true;
|
||||
} catch (error) {
|
||||
console.error('[iFlow] Token refresh failed:', error.message);
|
||||
|
|
|
|||
|
|
@ -12,7 +12,6 @@ import { getProviderModels } from '../provider-models.js';
|
|||
import { handleQwenOAuth } from '../../auth/oauth-handlers.js';
|
||||
import { configureAxiosProxy } from '../../utils/proxy-utils.js';
|
||||
import { isRetryableNetworkError } from '../../utils/common.js';
|
||||
import { CredentialCacheManager } from '../../utils/credential-cache-manager.js';
|
||||
|
||||
// --- Constants ---
|
||||
const QWEN_DIR = '.qwen';
|
||||
|
|
@ -389,18 +388,14 @@ export class QwenApiService {
|
|||
|
||||
async _cacheQwenCredentials(credentials) {
|
||||
const filePath = this._getQwenCachedCredentialPath();
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
// 使用内存锁替代文件锁，防止并发写入
|
||||
await credentialCache.withMemoryLock(`qwen-cache:${filePath}`, async () => {
|
||||
try {
|
||||
await fs.mkdir(path.dirname(filePath), { recursive: true });
|
||||
const credString = JSON.stringify(credentials, null, 2);
|
||||
await fs.writeFile(filePath, credString);
|
||||
console.log(`[Qwen Auth] Credentials cached to ${filePath}`);
|
||||
} catch (error) {
|
||||
console.error(`[Qwen Auth] Failed to cache credentials to ${filePath}: ${error.message}`);
|
||||
}
|
||||
});
|
||||
try {
|
||||
await fs.mkdir(path.dirname(filePath), { recursive: true });
|
||||
const credString = JSON.stringify(credentials, null, 2);
|
||||
await fs.writeFile(filePath, credString);
|
||||
console.log(`[Qwen Auth] Credentials cached to ${filePath}`);
|
||||
} catch (error) {
|
||||
console.error(`[Qwen Auth] Failed to cache credentials to ${filePath}: ${error.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
getCurrentEndpoint(resourceUrl) {
|
||||
|
|
@ -794,77 +789,39 @@ class SharedTokenManager {
|
|||
throw new TokenManagerError(TokenError.NO_REFRESH_TOKEN, 'No refresh token available');
|
||||
}
|
||||
|
||||
// 获取凭证缓存管理器
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'openai-qwen-oauth';
|
||||
|
||||
// 使用内存锁进行去重
|
||||
const dedupeKey = `qwen-token-refresh:${context.credentialFilePath}`;
|
||||
|
||||
try {
|
||||
const credentials = await credentialCache.withDeduplication(dedupeKey, async () => {
|
||||
await this.acquireLock(context);
|
||||
try {
|
||||
// 优先从全局缓存检查
|
||||
if (context.uuid && credentialCache.hasCredentials(providerType, context.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, context.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials && !forceRefresh && this.isTokenValid(cachedEntry.credentials)) {
|
||||
context.memoryCache.credentials = cachedEntry.credentials;
|
||||
qwenClient.setCredentials(cachedEntry.credentials);
|
||||
return cachedEntry.credentials;
|
||||
}
|
||||
}
|
||||
await this.acquireLock(context);
|
||||
try {
|
||||
await this.checkAndReloadIfNeeded(context);
|
||||
|
||||
await this.checkAndReloadIfNeeded(context);
|
||||
|
||||
if (!forceRefresh && context.memoryCache.credentials && this.isTokenValid(context.memoryCache.credentials)) {
|
||||
qwenClient.setCredentials(context.memoryCache.credentials);
|
||||
return context.memoryCache.credentials;
|
||||
}
|
||||
|
||||
const response = await qwenClient.refreshAccessToken();
|
||||
if (!response || isErrorResponse(response)) {
|
||||
throw new TokenManagerError(TokenError.REFRESH_FAILED, `Token refresh failed: ${response?.error}`);
|
||||
}
|
||||
if (!response.access_token) {
|
||||
throw new TokenManagerError(TokenError.REFRESH_FAILED, 'No access token in refresh response');
|
||||
}
|
||||
const newCredentials = {
|
||||
access_token: response.access_token,
|
||||
token_type: response.token_type,
|
||||
refresh_token: response.refresh_token || currentCredentials.refresh_token,
|
||||
resource_url: response.resource_url,
|
||||
expiry_date: Date.now() + response.expires_in * 1000,
|
||||
};
|
||||
|
||||
context.memoryCache.credentials = newCredentials;
|
||||
qwenClient.setCredentials(newCredentials);
|
||||
await this.saveCredentialsToFile(context, newCredentials);
|
||||
console.log('[Qwen Auth] Token refresh response: ok');
|
||||
return newCredentials;
|
||||
} finally {
|
||||
await this.releaseLock(context);
|
||||
if (!forceRefresh && context.memoryCache.credentials && this.isTokenValid(context.memoryCache.credentials)) {
|
||||
qwenClient.setCredentials(context.memoryCache.credentials);
|
||||
return context.memoryCache.credentials;
|
||||
}
|
||||
});
|
||||
|
||||
// 如果是等待其他请求完成的刷新，需要重新加载凭证
|
||||
if (!context.memoryCache.credentials || !this.isTokenValid(context.memoryCache.credentials)) {
|
||||
// 优先从全局缓存加载
|
||||
if (context.uuid && credentialCache.hasCredentials(providerType, context.uuid)) {
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, context.uuid);
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
context.memoryCache.credentials = cachedEntry.credentials;
|
||||
qwenClient.setCredentials(cachedEntry.credentials);
|
||||
}
|
||||
} else {
|
||||
await this.reloadCredentialsFromFile(context);
|
||||
if (context.memoryCache.credentials) {
|
||||
qwenClient.setCredentials(context.memoryCache.credentials);
|
||||
}
|
||||
const response = await qwenClient.refreshAccessToken();
|
||||
if (!response || isErrorResponse(response)) {
|
||||
throw new TokenManagerError(TokenError.REFRESH_FAILED, `Token refresh failed: ${response?.error}`);
|
||||
}
|
||||
if (!response.access_token) {
|
||||
throw new TokenManagerError(TokenError.REFRESH_FAILED, 'No access token in refresh response');
|
||||
}
|
||||
const newCredentials = {
|
||||
access_token: response.access_token,
|
||||
token_type: response.token_type,
|
||||
refresh_token: response.refresh_token || currentCredentials.refresh_token,
|
||||
resource_url: response.resource_url,
|
||||
expiry_date: Date.now() + response.expires_in * 1000,
|
||||
};
|
||||
|
||||
context.memoryCache.credentials = newCredentials;
|
||||
qwenClient.setCredentials(newCredentials);
|
||||
await this.saveCredentialsToFile(context, newCredentials);
|
||||
console.log('[Qwen Auth] Token refresh response: ok');
|
||||
return newCredentials;
|
||||
} finally {
|
||||
await this.releaseLock(context);
|
||||
}
|
||||
|
||||
return credentials;
|
||||
} catch (error) {
|
||||
if (error instanceof TokenManagerError) throw error;
|
||||
|
||||
|
|
@ -890,26 +847,14 @@ class SharedTokenManager {
|
|||
}
|
||||
|
||||
async saveCredentialsToFile(context, credentials) {
|
||||
// 优先更新内存缓存
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const providerType = 'openai-qwen-oauth';
|
||||
|
||||
if (context.uuid && credentialCache.hasCredentials(providerType, context.uuid)) {
|
||||
credentialCache.updateCredentials(providerType, context.uuid, credentials, context.credentialFilePath);
|
||||
console.info(`[Qwen Auth] Updated credentials in memory cache: ${context.uuid}`);
|
||||
// 同时更新本地内存缓存
|
||||
context.memoryCache.credentials = credentials;
|
||||
context.memoryCache.fileModTime = Date.now();
|
||||
return;
|
||||
}
|
||||
|
||||
// 回退到使用内存锁写入文件
|
||||
await credentialCache.withMemoryLock(`qwen-save:${context.credentialFilePath}`, async () => {
|
||||
try {
|
||||
await fs.mkdir(path.dirname(context.credentialFilePath), { recursive: true, mode: 0o700 });
|
||||
await fs.writeFile(context.credentialFilePath, JSON.stringify(credentials, null, 2), { mode: 0o600 });
|
||||
const stats = await fs.stat(context.credentialFilePath);
|
||||
context.memoryCache.fileModTime = stats.mtimeMs;
|
||||
});
|
||||
} catch (error) {
|
||||
console.error(`[Qwen Auth] Failed to save credentials to ${context.credentialFilePath}: ${error.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
isTokenValid(credentials) {
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ import * as fs from 'fs'; // Import fs module
|
|||
import { getServiceAdapter } from './adapter.js';
|
||||
import { MODEL_PROVIDER, getProtocolPrefix } from '../utils/common.js';
|
||||
import { getProviderModels } from './provider-models.js';
|
||||
import { CredentialCacheManager } from '../utils/credential-cache-manager.js';
|
||||
import axios from 'axios';
|
||||
|
||||
/**
|
||||
|
|
@ -884,45 +883,6 @@ export class ProviderPoolManager {
|
|||
};
|
||||
}
|
||||
|
||||
// ========== 快速预检：从内存缓存检查 OAuth 凭证状态 ==========
|
||||
// 对于 OAuth 类型的 provider，先检查 token 是否过期，避免阻塞
|
||||
// 注意：只有在开启自动刷新 token (CRON_REFRESH_TOKEN) 时才执行快速预检
|
||||
// 因为如果没有自动刷新机制，缓存中的 token 状态可能不准确
|
||||
const oauthProviderTypes = ['claude-kiro-oauth', 'gemini-cli-oauth', 'gemini-antigravity', 'openai-qwen-oauth', 'openai-iflow-oauth', 'claude-orchids-oauth'];
|
||||
const cronRefreshTokenEnabled = this.globalConfig?.CRON_REFRESH_TOKEN === true;
|
||||
|
||||
if (cronRefreshTokenEnabled && oauthProviderTypes.includes(providerType) && providerConfig.uuid) {
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
const cachedEntry = credentialCache.getCredentials(providerType, providerConfig.uuid);
|
||||
|
||||
if (cachedEntry && cachedEntry.credentials) {
|
||||
const { expiresAt, accessToken } = cachedEntry.credentials;
|
||||
|
||||
// 检查 token 是否存在
|
||||
if (!accessToken) {
|
||||
this._log('warn', `Health check fast-fail for ${providerConfig.uuid}: No access token in cache`);
|
||||
return { success: false, modelName, errorMessage: 'No access token available' };
|
||||
}
|
||||
|
||||
// 检查 token 是否已过期（30秒缓冲）
|
||||
if (expiresAt) {
|
||||
const expirationTime = new Date(expiresAt).getTime();
|
||||
const now = Date.now();
|
||||
const bufferMs = 30 * 1000;
|
||||
|
||||
if (expirationTime <= now + bufferMs) {
|
||||
this._log('warn', `Health check fast-fail for ${providerConfig.uuid}: Token expired`);
|
||||
return { success: false, modelName, errorMessage: 'Token expired' };
|
||||
}
|
||||
}
|
||||
|
||||
this._log('debug', `Health check pre-check passed for ${providerConfig.uuid}: Token valid in cache`);
|
||||
}
|
||||
// 注意：如果缓存中没有凭证，不要直接返回失败
|
||||
// 让后续的实际健康检查去尝试初始化和加载凭证
|
||||
// 这样可以支持刚重置健康状态或新添加的 provider
|
||||
}
|
||||
|
||||
// ========== 实际 API 健康检查（带超时保护）==========
|
||||
const tempConfig = {
|
||||
...providerConfig,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
import { getServiceAdapter, serviceInstances } from '../providers/adapter.js';
|
||||
import { ProviderPoolManager } from '../providers/provider-pool-manager.js';
|
||||
import { CredentialCacheManager } from '../utils/credential-cache-manager.js';
|
||||
import deepmerge from 'deepmerge';
|
||||
import * as fs from 'fs';
|
||||
import { promises as pfs } from 'fs';
|
||||
|
|
@ -176,62 +175,6 @@ export async function initApiService(config) {
|
|||
});
|
||||
console.log('[Initialization] ProviderPoolManager initialized with configured pools.');
|
||||
// 健康检查将在服务器完全启动后执行
|
||||
|
||||
// 初始化凭证缓存管理器
|
||||
const credentialCache = CredentialCacheManager.getInstance();
|
||||
|
||||
// 获取单实例锁,防止多实例并发运行
|
||||
try {
|
||||
await credentialCache.acquireInstanceLock();
|
||||
} catch (lockError) {
|
||||
console.error('[Initialization] Failed to acquire instance lock:', lockError.message);
|
||||
console.error('[Initialization] Please ensure no other instance is running.');
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
await credentialCache.preloadAllCredentials(config.providerPools);
|
||||
credentialCache.startPeriodicSync(config.CREDENTIAL_SYNC_INTERVAL || 5000);
|
||||
|
||||
// 注册进程退出钩子
|
||||
const shutdownHandler = async () => {
|
||||
await credentialCache.shutdown();
|
||||
};
|
||||
process.on('beforeExit', shutdownHandler);
|
||||
process.on('SIGINT', async () => {
|
||||
await shutdownHandler();
|
||||
process.exit(0);
|
||||
});
|
||||
process.on('SIGTERM', async () => {
|
||||
await shutdownHandler();
|
||||
process.exit(0);
|
||||
});
|
||||
|
||||
// 处理未捕获的异常,进行紧急同步
|
||||
process.on('uncaughtException', async (error) => {
|
||||
console.error('[FATAL] Uncaught exception:', error);
|
||||
console.log('[CredentialCache] Attempting emergency sync before crash...');
|
||||
try {
|
||||
await credentialCache.syncToFile();
|
||||
console.log('[CredentialCache] Emergency sync completed');
|
||||
} catch (syncError) {
|
||||
console.error('[CredentialCache] Emergency sync failed:', syncError.message);
|
||||
}
|
||||
await credentialCache.shutdown();
|
||||
process.exit(1);
|
||||
});
|
||||
|
||||
process.on('unhandledRejection', async (reason, promise) => {
|
||||
console.error('[FATAL] Unhandled rejection at:', promise, 'reason:', reason);
|
||||
console.log('[CredentialCache] Attempting emergency sync...');
|
||||
try {
|
||||
await credentialCache.syncToFile();
|
||||
console.log('[CredentialCache] Emergency sync completed');
|
||||
} catch (syncError) {
|
||||
console.error('[CredentialCache] Emergency sync failed:', syncError.message);
|
||||
}
|
||||
});
|
||||
|
||||
console.log('[Initialization] CredentialCacheManager initialized.');
|
||||
} else {
|
||||
console.log('[Initialization] No provider pools configured. Using single provider mode.');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,925 +0,0 @@
|
|||
/**
|
||||
* 凭证缓存管理器
|
||||
* 纯内存操作,使用内存锁保证并发安全,定时同步到文件
|
||||
* 优先考虑并发性能,支持单实例部署
|
||||
*/
|
||||
|
||||
import { promises as fs } from 'fs';
|
||||
import * as path from 'path';
|
||||
import { PROVIDER_MAPPINGS } from './provider-utils.js';
|
||||
import * as os from 'os';
|
||||
|
||||
/**
|
||||
* 凭证条目结构
|
||||
* @typedef {Object} CredentialEntry
|
||||
* @property {string} providerType - 提供商类型
|
||||
* @property {string} uuid - 节点唯一标识
|
||||
* @property {string} credPath - 凭证文件路径
|
||||
* @property {Object} credentials - 凭证数据
|
||||
* @property {number} lastModified - 最后修改时间戳
|
||||
* @property {number} lastAccessed - 最后访问时间戳
|
||||
* @property {boolean} isDirty - 是否有未同步的修改
|
||||
* @property {number} retryCount - 同步失败重试次数
|
||||
*/
|
||||
|
||||
export class CredentialCacheManager {
|
||||
static instance = null;
|
||||
|
||||
constructor() {
|
||||
// 凭证缓存: Map<cacheKey, CredentialEntry>
|
||||
// cacheKey = `${providerType}:${uuid}`
|
||||
this.credentialCache = new Map();
|
||||
|
||||
// Promise链锁: Map<lockKey, Promise>
|
||||
this.lockChains = new Map();
|
||||
|
||||
// 脏数据标记（需要同步到文件的 cacheKey）
|
||||
this.dirtyKeys = new Set();
|
||||
|
||||
// 同步定时器
|
||||
this.syncTimer = null;
|
||||
this.syncIntervalMs = 5000; // 默认5秒
|
||||
|
||||
// 是否已初始化
|
||||
this.isInitialized = false;
|
||||
|
||||
// 是否正在同步
|
||||
this.isSyncing = false;
|
||||
|
||||
// 最大重试次数
|
||||
this.maxRetries = 5;
|
||||
|
||||
// 最大脏数据条目数
|
||||
this.maxDirtyKeys = 1000;
|
||||
|
||||
// 死信队列 - 存储同步失败的凭证
|
||||
this.deadLetterQueue = new Map();
|
||||
|
||||
// 实例锁文件句柄
|
||||
this.instanceLockRelease = null;
|
||||
|
||||
// 提供商类型到凭证路径键的映射
|
||||
this.providerCredPathKeys = {};
|
||||
for (const mapping of PROVIDER_MAPPINGS) {
|
||||
this.providerCredPathKeys[mapping.providerType] = mapping.credPathKey;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取单例实例
|
||||
* @returns {CredentialCacheManager}
|
||||
*/
|
||||
static getInstance() {
|
||||
if (!CredentialCacheManager.instance) {
|
||||
CredentialCacheManager.instance = new CredentialCacheManager();
|
||||
}
|
||||
return CredentialCacheManager.instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* 生成缓存键
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {string} uuid - 节点UUID
|
||||
* @returns {string}
|
||||
*/
|
||||
_getCacheKey(providerType, uuid) {
|
||||
return `${providerType}:${uuid}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取单实例锁,防止多实例并发运行
|
||||
* @throws {Error} 如果已有实例在运行
|
||||
*/
|
||||
async acquireInstanceLock() {
|
||||
const lockPath = path.join(os.tmpdir(), 'credential-cache.lock');
|
||||
const pidPath = path.join(os.tmpdir(), 'credential-cache.pid');
|
||||
|
||||
try {
|
||||
// 尝试读取已存在的 PID 文件
|
||||
try {
|
||||
const existingPid = await fs.readFile(pidPath, 'utf8');
|
||||
const pid = parseInt(existingPid.trim(), 10);
|
||||
|
||||
// 如果是当前进程的 PID，说明是配置重载，直接返回（已持有锁）
|
||||
if (pid === process.pid) {
|
||||
console.log(`[CredentialCache] Instance lock already held by current process (PID: ${pid})`);
|
||||
return;
|
||||
}
|
||||
|
||||
// 检查进程是否还在运行
|
||||
try {
|
||||
process.kill(pid, 0); // 0 信号仅检查进程存在性
|
||||
// 如果进程存在，检查是否是当前进程
|
||||
throw new Error(`[CredentialCache] Another instance is running (PID: ${pid}). Please stop it first.`);
|
||||
} catch (killError) {
|
||||
if (killError.code === 'ESRCH') {
|
||||
// 进程已死亡,可以继续
|
||||
console.log(`[CredentialCache] Stale lock detected (PID: ${pid}), cleaning up...`);
|
||||
} else {
|
||||
throw killError;
|
||||
}
|
||||
}
|
||||
} catch (readError) {
|
||||
if (readError.code !== 'ENOENT') {
|
||||
throw readError;
|
||||
}
|
||||
// PID 文件不存在,可以继续
|
||||
}
|
||||
|
||||
// 写入当前进程 PID
|
||||
await fs.writeFile(pidPath, String(process.pid), 'utf8');
|
||||
console.log(`[CredentialCache] Instance lock acquired (PID: ${process.pid})`);
|
||||
|
||||
// 注册清理钩子
|
||||
this.instanceLockRelease = async () => {
|
||||
try {
|
||||
await fs.unlink(pidPath);
|
||||
console.log('[CredentialCache] Instance lock released');
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.warn(`[CredentialCache] Failed to release lock: ${error.message}`);
|
||||
}
|
||||
}
|
||||
};
|
||||
} catch (error) {
|
||||
console.error(`[CredentialCache] Failed to acquire instance lock: ${error.message}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 预加载所有凭证到内存
|
||||
* @param {Object} providerPools - 提供商池配置
|
||||
*/
|
||||
async preloadAllCredentials(providerPools) {
|
||||
if (!providerPools || typeof providerPools !== 'object') {
|
||||
console.log('[CredentialCache] No provider pools to preload');
|
||||
return;
|
||||
}
|
||||
|
||||
let loadedCount = 0;
|
||||
let failedCount = 0;
|
||||
let skippedCount = 0;
|
||||
|
||||
for (const [providerType, providers] of Object.entries(providerPools)) {
|
||||
if (!Array.isArray(providers)) continue;
|
||||
|
||||
const credPathKey = this.providerCredPathKeys[providerType];
|
||||
if (!credPathKey) {
|
||||
console.warn(`[CredentialCache] Unknown provider type: ${providerType}`);
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const providerConfig of providers) {
|
||||
const uuid = providerConfig.uuid;
|
||||
const credPath = providerConfig[credPathKey];
|
||||
const customName = providerConfig.customName || uuid;
|
||||
|
||||
if (!uuid || !credPath) {
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const credentials = await this._loadCredentialsFromFile(credPath);
|
||||
if (credentials) {
|
||||
// 预检查：验证凭证是否有效（至少有 token）
|
||||
const hasValidToken = this._validateCredentials(providerType, credentials);
|
||||
if (!hasValidToken) {
|
||||
skippedCount++;
|
||||
console.warn(`[CredentialCache] Skipping ${providerType}:${customName} - no valid token found in credentials`);
|
||||
continue;
|
||||
}
|
||||
|
||||
const cacheKey = this._getCacheKey(providerType, uuid);
|
||||
this.credentialCache.set(cacheKey, {
|
||||
providerType,
|
||||
uuid,
|
||||
credPath,
|
||||
credentials,
|
||||
lastModified: Date.now(),
|
||||
lastAccessed: Date.now(),
|
||||
isDirty: false,
|
||||
retryCount: 0
|
||||
});
|
||||
loadedCount++;
|
||||
} else {
|
||||
skippedCount++;
|
||||
console.warn(`[CredentialCache] Skipping ${providerType}:${customName} - credentials file empty or unreadable`);
|
||||
}
|
||||
} catch (error) {
|
||||
failedCount++;
|
||||
console.warn(`[CredentialCache] Failed to load credentials for ${providerType}:${customName}: ${error.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
this.isInitialized = true;
|
||||
console.log(`[CredentialCache] Preloaded ${loadedCount} credentials (${failedCount} failed, ${skippedCount} skipped)`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证凭证是否有效
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {Object} credentials - 凭证对象
|
||||
* @returns {boolean}
|
||||
*/
|
||||
_validateCredentials(providerType, credentials) {
|
||||
if (!credentials || typeof credentials !== 'object') {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 根据提供商类型检查必要的 token 字段
|
||||
switch (providerType) {
|
||||
case 'claude-kiro-oauth':
|
||||
// Kiro 需要 refreshToken 或 accessToken
|
||||
return !!(credentials.refreshToken || credentials.accessToken);
|
||||
|
||||
case 'gemini-cli-oauth':
|
||||
case 'gemini-antigravity':
|
||||
// Gemini 需要 refresh_token 或 access_token
|
||||
return !!(credentials.refresh_token || credentials.access_token);
|
||||
|
||||
case 'openai-qwen-oauth':
|
||||
// Qwen 需要 refresh_token 或 access_token
|
||||
return !!(credentials.refresh_token || credentials.access_token);
|
||||
|
||||
case 'openai-iflow':
|
||||
// iFlow 需要 refresh_token 或 access_token
|
||||
return !!(credentials.refresh_token || credentials.access_token);
|
||||
|
||||
case 'claude-orchids-oauth':
|
||||
// Orchids 需要 sessionKey 或 accessToken
|
||||
return !!(credentials.sessionKey || credentials.accessToken);
|
||||
|
||||
default:
|
||||
// 默认：只要有任何 token 相关字段就认为有效
|
||||
return !!(
|
||||
credentials.refreshToken || credentials.accessToken ||
|
||||
credentials.refresh_token || credentials.access_token ||
|
||||
credentials.sessionKey || credentials.apiKey
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 原子写入文件 (temp + rename 模式)
|
||||
* @param {string} filePath - 目标文件路径
|
||||
* @param {string} data - 要写入的数据
|
||||
*/
|
||||
async _atomicWriteFile(filePath, data) {
|
||||
const tmpPath = `${filePath}.tmp.${Date.now()}.${process.pid}`;
|
||||
|
||||
try {
|
||||
// 1. 写入临时文件
|
||||
await fs.writeFile(tmpPath, data, 'utf8');
|
||||
|
||||
// 2. fsync 确保落盘 (需要文件句柄)
|
||||
const fileHandle = await fs.open(tmpPath, 'r+');
|
||||
try {
|
||||
await fileHandle.sync();
|
||||
} finally {
|
||||
await fileHandle.close();
|
||||
}
|
||||
|
||||
// 3. 原子重命名
|
||||
await fs.rename(tmpPath, filePath);
|
||||
|
||||
return true;
|
||||
} catch (error) {
|
||||
// 清理临时文件
|
||||
try {
|
||||
await fs.unlink(tmpPath);
|
||||
} catch (unlinkError) {
|
||||
// 忽略清理失败
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* 从文件加载凭证（仅用于初始导入和手动重载）
|
||||
* 支持从损坏文件的备份恢复
|
||||
* @param {string} filePath - 文件路径
|
||||
* @returns {Promise<Object|null>}
|
||||
*/
|
||||
async _loadCredentialsFromFile(filePath) {
|
||||
try {
|
||||
// 处理相对路径
|
||||
const absolutePath = path.isAbsolute(filePath)
|
||||
? filePath
|
||||
: path.join(process.cwd(), filePath);
|
||||
|
||||
const content = await fs.readFile(absolutePath, 'utf8');
|
||||
try {
|
||||
return JSON.parse(content);
|
||||
} catch (parseError) {
|
||||
console.warn(`[CredentialCache] JSON parse failed for ${filePath}, attempting field extraction...`);
|
||||
// 尝试从损坏的 JSON 中提取关键字段
|
||||
const extracted = this._extractCredentialsFromCorruptedJson(content);
|
||||
if (extracted) {
|
||||
console.info(`[CredentialCache] Field extraction successful for ${filePath}`);
|
||||
return extracted;
|
||||
}
|
||||
console.error(`[CredentialCache] All recovery methods failed for ${filePath}: ${parseError.message}`);
|
||||
return null;
|
||||
}
|
||||
} catch (error) {
|
||||
if (error.code === 'ENOENT') {
|
||||
return null;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 从损坏的 JSON 中提取关键凭证字段
|
||||
* @param {string} content - 文件内容
|
||||
* @returns {Object|null}
|
||||
*/
|
||||
_extractCredentialsFromCorruptedJson(content) {
|
||||
const extracted = {};
|
||||
|
||||
// 定义需要提取的关键字段及其正则模式
|
||||
const fieldPatterns = {
|
||||
refreshToken: /"refreshToken"\s*:\s*"([^"]+)"/,
|
||||
accessToken: /"accessToken"\s*:\s*"([^"]+)"/,
|
||||
clientId: /"clientId"\s*:\s*"([^"]+)"/,
|
||||
clientSecret: /"clientSecret"\s*:\s*"([^"]+)"/,
|
||||
profileArn: /"profileArn"\s*:\s*"([^"]+)"/,
|
||||
region: /"region"\s*:\s*"([^"]+)"/,
|
||||
authMethod: /"authMethod"\s*:\s*"([^"]+)"/,
|
||||
expiresAt: /"expiresAt"\s*:\s*"([^"]+)"/,
|
||||
startUrl: /"startUrl"\s*:\s*"([^"]+)"/,
|
||||
// Gemini/Qwen 相关字段
|
||||
refresh_token: /"refresh_token"\s*:\s*"([^"]+)"/,
|
||||
access_token: /"access_token"\s*:\s*"([^"]+)"/,
|
||||
client_id: /"client_id"\s*:\s*"([^"]+)"/,
|
||||
client_secret: /"client_secret"\s*:\s*"([^"]+)"/,
|
||||
};
|
||||
|
||||
for (const [field, pattern] of Object.entries(fieldPatterns)) {
|
||||
const match = content.match(pattern);
|
||||
if (match && match[1]) {
|
||||
extracted[field] = match[1];
|
||||
}
|
||||
}
|
||||
|
||||
// 至少需要某种 token 才算有效
|
||||
if (extracted.refreshToken || extracted.accessToken ||
|
||||
extracted.refresh_token || extracted.access_token) {
|
||||
console.info(`[CredentialCache] Extracted ${Object.keys(extracted).length} fields: ${Object.keys(extracted).join(', ')}`);
|
||||
return extracted;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取凭证（从内存）
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {string} uuid - 节点UUID
|
||||
* @returns {CredentialEntry|null}
|
||||
*/
|
||||
getCredentials(providerType, uuid) {
|
||||
const cacheKey = this._getCacheKey(providerType, uuid);
|
||||
const entry = this.credentialCache.get(cacheKey);
|
||||
|
||||
if (entry) {
|
||||
entry.lastAccessed = Date.now();
|
||||
return entry;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查凭证是否存在于缓存中
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {string} uuid - 节点UUID
|
||||
* @returns {boolean}
|
||||
*/
|
||||
hasCredentials(providerType, uuid) {
|
||||
const cacheKey = this._getCacheKey(providerType, uuid);
|
||||
return this.credentialCache.has(cacheKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新凭证（仅更新内存，标记为dirty）
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {string} uuid - 节点UUID
|
||||
* @param {Object} newCredentials - 新凭证数据
|
||||
* @param {string} [credPath] - 凭证文件路径（可选，用于新建条目）
|
||||
*/
|
||||
updateCredentials(providerType, uuid, newCredentials, credPath = null) {
|
||||
const cacheKey = this._getCacheKey(providerType, uuid);
|
||||
let entry = this.credentialCache.get(cacheKey);
|
||||
|
||||
if (entry) {
|
||||
// 更新现有条目
|
||||
entry.credentials = newCredentials;
|
||||
entry.lastModified = Date.now();
|
||||
entry.isDirty = true;
|
||||
entry.retryCount = 0; // 重置重试计数
|
||||
} else if (credPath) {
|
||||
// 创建新条目
|
||||
entry = {
|
||||
providerType,
|
||||
uuid,
|
||||
credPath,
|
||||
credentials: newCredentials,
|
||||
lastModified: Date.now(),
|
||||
lastAccessed: Date.now(),
|
||||
isDirty: true,
|
||||
retryCount: 0
|
||||
};
|
||||
this.credentialCache.set(cacheKey, entry);
|
||||
} else {
|
||||
console.warn(`[CredentialCache] Cannot update non-existent entry without credPath: ${cacheKey}`);
|
||||
return;
|
||||
}
|
||||
|
||||
// 标记为需要同步
|
||||
this.dirtyKeys.add(cacheKey);
|
||||
|
||||
// 检查脏数据是否过多
|
||||
if (this.dirtyKeys.size > this.maxDirtyKeys) {
|
||||
console.warn(`[CredentialCache] Dirty keys exceeded ${this.maxDirtyKeys}, triggering immediate sync`);
|
||||
this.syncToFile().catch(err => console.error('[CredentialCache] Emergency sync failed:', err.message));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除凭证（从内存中删除，同时删除文件）
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {string} uuid - 节点UUID
|
||||
*/
|
||||
async deleteCredentials(providerType, uuid) {
|
||||
const cacheKey = this._getCacheKey(providerType, uuid);
|
||||
const entry = this.credentialCache.get(cacheKey);
|
||||
|
||||
if (!entry) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 从内存中删除
|
||||
this.credentialCache.delete(cacheKey);
|
||||
this.dirtyKeys.delete(cacheKey);
|
||||
|
||||
// 删除文件
|
||||
if (entry.credPath) {
|
||||
try {
|
||||
const absolutePath = path.isAbsolute(entry.credPath)
|
||||
? entry.credPath
|
||||
: path.join(process.cwd(), entry.credPath);
|
||||
await fs.unlink(absolutePath);
|
||||
console.log(`[CredentialCache] Deleted credential file: ${entry.credPath}`);
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.error(`[CredentialCache] Failed to delete ${entry.credPath}: ${error.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Promise链式内存锁 - 保证串行执行
|
||||
* 使用 Promise 构造器模式避免 Read-Check-Write 竞态
|
||||
* @param {string} key - 锁的唯一标识
|
||||
* @param {Function} operation - 要执行的操作
|
||||
* @returns {Promise<T>}
|
||||
*/
|
||||
async withMemoryLock(key, operation) {
|
||||
// 创建新的 Promise 用于链接后续操作
|
||||
let resolveNext, rejectNext;
|
||||
const nextPromise = new Promise((resolve, reject) => {
|
||||
resolveNext = resolve;
|
||||
rejectNext = reject;
|
||||
});
|
||||
|
||||
// 原子化获取前序链并立即更新 Map (避免竞态)
|
||||
const prevChain = this.lockChains.get(key);
|
||||
this.lockChains.set(key, nextPromise);
|
||||
|
||||
// 链接到前序 Promise 执行操作
|
||||
const executeOperation = async () => {
|
||||
try {
|
||||
// 等待前序操作完成（忽略前序的错误，继续执行当前操作）
|
||||
if (prevChain) {
|
||||
await prevChain.catch(() => {});
|
||||
}
|
||||
const result = await operation();
|
||||
resolveNext(result);
|
||||
} catch (error) {
|
||||
rejectNext(error);
|
||||
} finally {
|
||||
// 只有当前 Promise 还在 Map 中时才删除
|
||||
if (this.lockChains.get(key) === nextPromise) {
|
||||
this.lockChains.delete(key);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// 立即开始执行（不阻塞）
|
||||
executeOperation();
|
||||
|
||||
return nextPromise;
|
||||
}
|
||||
|
||||
/**
|
||||
* 去重执行 - 多个并发请求共享同一个操作结果
|
||||
* @param {string} key - 去重键
|
||||
* @param {Function} operation - 要执行的操作
|
||||
* @returns {Promise<T>}
|
||||
*/
|
||||
async withDeduplication(key, operation) {
|
||||
const dedupeKey = `dedupe:${key}`;
|
||||
|
||||
// 检查是否已有正在执行的操作
|
||||
const existingPromise = this.lockChains.get(dedupeKey);
|
||||
if (existingPromise) {
|
||||
// 直接等待现有操作完成并返回结果
|
||||
return existingPromise;
|
||||
}
|
||||
|
||||
// 创建新操作 Promise 并立即存储（避免竞态）
|
||||
const operationPromise = (async () => {
|
||||
try {
|
||||
return await operation();
|
||||
} finally {
|
||||
// 操作完成后清理
|
||||
if (this.lockChains.get(dedupeKey) === operationPromise) {
|
||||
this.lockChains.delete(dedupeKey);
|
||||
}
|
||||
}
|
||||
})();
|
||||
|
||||
// 立即存储，确保后续请求能看到
|
||||
this.lockChains.set(dedupeKey, operationPromise);
|
||||
|
||||
return operationPromise;
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动定时同步
|
||||
* @param {number} intervalMs - 同步间隔（毫秒）
|
||||
*/
|
||||
startPeriodicSync(intervalMs = 5000) {
|
||||
this.syncIntervalMs = intervalMs;
|
||||
|
||||
if (this.syncTimer) {
|
||||
clearInterval(this.syncTimer);
|
||||
}
|
||||
|
||||
this.syncTimer = setInterval(() => {
|
||||
this.syncToFile().catch(error => {
|
||||
console.error('[CredentialCache] Periodic sync failed:', error.message);
|
||||
});
|
||||
}, this.syncIntervalMs);
|
||||
|
||||
// 确保定时器不阻止进程退出
|
||||
if (this.syncTimer.unref) {
|
||||
this.syncTimer.unref();
|
||||
}
|
||||
|
||||
console.log(`[CredentialCache] Started periodic sync (interval: ${intervalMs}ms)`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 停止定时同步
|
||||
*/
|
||||
stopPeriodicSync() {
|
||||
if (this.syncTimer) {
|
||||
clearInterval(this.syncTimer);
|
||||
this.syncTimer = null;
|
||||
console.log('[CredentialCache] Stopped periodic sync');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步脏数据到文件（纯内存操作，不使用文件锁）
|
||||
*/
|
||||
async syncToFile() {
|
||||
if (this.dirtyKeys.size === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.isSyncing) {
|
||||
return; // 防止重入
|
||||
}
|
||||
|
||||
this.isSyncing = true;
|
||||
|
||||
// 复制脏数据集合（不立即清空，同步成功后再清理）
|
||||
const keysToSync = Array.from(this.dirtyKeys);
|
||||
|
||||
console.log(`[CredentialCache] Syncing ${keysToSync.length} credential(s) to file...`);
|
||||
|
||||
let successCount = 0;
|
||||
let failedCount = 0;
|
||||
const successKeys = new Set();
|
||||
|
||||
// 并发写入所有文件（提高性能）
|
||||
await Promise.allSettled(
|
||||
keysToSync.map(async (cacheKey) => {
|
||||
const entry = this.credentialCache.get(cacheKey);
|
||||
if (!entry || !entry.credPath) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 检查重试次数
|
||||
if (entry.retryCount >= this.maxRetries) {
|
||||
// 达到最大重试次数,移入死信队列
|
||||
console.error(`[CredentialCache] Max retries exceeded for ${cacheKey}, moving to dead letter queue`);
|
||||
|
||||
this.deadLetterQueue.set(cacheKey, {
|
||||
entry: JSON.parse(JSON.stringify(entry)), // 深拷贝
|
||||
failureReason: entry.lastError || 'Max retries exceeded',
|
||||
firstFailureTime: entry.firstFailureTime || Date.now(),
|
||||
timestamp: Date.now()
|
||||
});
|
||||
|
||||
// 尝试导出到紧急备份
|
||||
try {
|
||||
const emergencyDir = path.join(process.cwd(), 'emergency_backup');
|
||||
await fs.mkdir(emergencyDir, { recursive: true });
|
||||
const emergencyPath = path.join(emergencyDir, `${cacheKey.replace(/:/g, '_')}.json`);
|
||||
await fs.writeFile(emergencyPath, JSON.stringify(entry.credentials, null, 2), 'utf8');
|
||||
console.log(`[CredentialCache] Credential backed up to: ${emergencyPath}`);
|
||||
} catch (backupError) {
|
||||
console.error(`[CredentialCache] Failed to backup credential: ${backupError.message}`);
|
||||
}
|
||||
|
||||
this.dirtyKeys.delete(cacheKey);
|
||||
entry.isDirty = false;
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// 处理相对路径
|
||||
const absolutePath = path.isAbsolute(entry.credPath)
|
||||
? entry.credPath
|
||||
: path.join(process.cwd(), entry.credPath);
|
||||
|
||||
// 确保目录存在
|
||||
const dir = path.dirname(absolutePath);
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
|
||||
// 使用原子写入
|
||||
await this._atomicWriteFile(
|
||||
absolutePath,
|
||||
JSON.stringify(entry.credentials, null, 2)
|
||||
);
|
||||
|
||||
entry.isDirty = false;
|
||||
entry.retryCount = 0;
|
||||
entry.lastError = null;
|
||||
entry.firstFailureTime = null;
|
||||
successKeys.add(cacheKey);
|
||||
successCount++;
|
||||
} catch (error) {
|
||||
// 同步失败，增加重试计数并记录错误
|
||||
entry.retryCount++;
|
||||
entry.lastError = error.message;
|
||||
if (!entry.firstFailureTime) {
|
||||
entry.firstFailureTime = Date.now();
|
||||
}
|
||||
failedCount++;
|
||||
console.error(`[CredentialCache] Failed to sync ${entry.credPath} (retry ${entry.retryCount}/${this.maxRetries}): ${error.message}`);
|
||||
}
|
||||
})
|
||||
);
|
||||
|
||||
// 从脏数据集合中移除成功同步的键
|
||||
for (const key of successKeys) {
|
||||
this.dirtyKeys.delete(key);
|
||||
}
|
||||
|
||||
this.isSyncing = false;
|
||||
|
||||
if (successCount > 0 || failedCount > 0) {
|
||||
console.log(`[CredentialCache] Sync completed: ${successCount} success, ${failedCount} failed, ${this.dirtyKeys.size} pending`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 立即同步指定凭证到文件
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {string} uuid - 节点UUID
|
||||
*/
|
||||
async syncCredentialToFile(providerType, uuid) {
|
||||
const cacheKey = this._getCacheKey(providerType, uuid);
|
||||
const entry = this.credentialCache.get(cacheKey);
|
||||
|
||||
if (!entry || !entry.credPath) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const absolutePath = path.isAbsolute(entry.credPath)
|
||||
? entry.credPath
|
||||
: path.join(process.cwd(), entry.credPath);
|
||||
|
||||
const dir = path.dirname(absolutePath);
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
|
||||
// 使用原子写入
|
||||
await this._atomicWriteFile(
|
||||
absolutePath,
|
||||
JSON.stringify(entry.credentials, null, 2)
|
||||
);
|
||||
|
||||
entry.isDirty = false;
|
||||
entry.retryCount = 0;
|
||||
entry.lastError = null;
|
||||
entry.firstFailureTime = null;
|
||||
this.dirtyKeys.delete(cacheKey);
|
||||
} catch (error) {
|
||||
console.error(`[CredentialCache] Failed to sync ${entry.credPath}: ${error.message}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭时同步所有数据（带超时）
|
||||
*/
|
||||
async shutdown() {
|
||||
console.log('[CredentialCache] Shutting down...');
|
||||
|
||||
// 停止定时同步
|
||||
this.stopPeriodicSync();
|
||||
|
||||
// 同步所有脏数据（带动态超时）
|
||||
if (this.dirtyKeys.size > 0) {
|
||||
console.log(`[CredentialCache] Syncing ${this.dirtyKeys.size} dirty credential(s) before shutdown...`);
|
||||
|
||||
// 根据脏数据量动态调整超时时间
|
||||
const timeoutMs = Math.max(10000, this.dirtyKeys.size * 50 + 5000);
|
||||
const timeoutPromise = new Promise((_, reject) =>
|
||||
setTimeout(() => reject(new Error('Shutdown sync timeout')), timeoutMs)
|
||||
);
|
||||
|
||||
try {
|
||||
await Promise.race([
|
||||
this.syncToFile(),
|
||||
timeoutPromise
|
||||
]);
|
||||
} catch (error) {
|
||||
console.error(`[CredentialCache] Shutdown sync failed: ${error.message}`);
|
||||
console.error(`[CredentialCache] ${this.dirtyKeys.size} credentials NOT saved`);
|
||||
|
||||
// 尝试紧急备份未保存的凭证
|
||||
if (this.dirtyKeys.size > 0) {
|
||||
console.log('[CredentialCache] Attempting emergency backup...');
|
||||
try {
|
||||
const emergencyDir = path.join(process.cwd(), 'emergency_backup');
|
||||
await fs.mkdir(emergencyDir, { recursive: true });
|
||||
for (const cacheKey of this.dirtyKeys) {
|
||||
const entry = this.credentialCache.get(cacheKey);
|
||||
if (entry && entry.credentials) {
|
||||
const emergencyPath = path.join(emergencyDir, `${cacheKey.replace(/:/g, '_')}.json`);
|
||||
await fs.writeFile(emergencyPath, JSON.stringify(entry.credentials, null, 2), 'utf8');
|
||||
}
|
||||
}
|
||||
console.log(`[CredentialCache] Emergency backup completed to: ${emergencyDir}`);
|
||||
} catch (backupError) {
|
||||
console.error(`[CredentialCache] Emergency backup failed: ${backupError.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 释放实例锁
|
||||
if (this.instanceLockRelease) {
|
||||
await this.instanceLockRelease();
|
||||
}
|
||||
|
||||
// 输出死信队列状态
|
||||
if (this.deadLetterQueue.size > 0) {
|
||||
console.warn(`[CredentialCache] ${this.deadLetterQueue.size} credential(s) in dead letter queue`);
|
||||
}
|
||||
|
||||
console.log('[CredentialCache] Shutdown complete');
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取缓存统计信息
|
||||
* @returns {Object}
|
||||
*/
|
||||
getStats() {
|
||||
const stats = {
|
||||
totalEntries: this.credentialCache.size,
|
||||
dirtyEntries: this.dirtyKeys.size,
|
||||
activeLocks: this.lockChains.size,
|
||||
deadLetterQueueSize: this.deadLetterQueue.size,
|
||||
isInitialized: this.isInitialized,
|
||||
isSyncing: this.isSyncing,
|
||||
syncInterval: this.syncIntervalMs,
|
||||
byProvider: {}
|
||||
};
|
||||
|
||||
for (const [cacheKey, entry] of this.credentialCache) {
|
||||
const providerType = entry.providerType;
|
||||
if (!stats.byProvider[providerType]) {
|
||||
stats.byProvider[providerType] = {
|
||||
count: 0,
|
||||
dirty: 0,
|
||||
maxRetries: 0
|
||||
};
|
||||
}
|
||||
stats.byProvider[providerType].count++;
|
||||
if (entry.isDirty) {
|
||||
stats.byProvider[providerType].dirty++;
|
||||
stats.byProvider[providerType].maxRetries = Math.max(
|
||||
stats.byProvider[providerType].maxRetries,
|
||||
entry.retryCount
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return stats;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取死信队列内容
|
||||
* @returns {Array}
|
||||
*/
|
||||
getDeadLetterQueue() {
|
||||
return Array.from(this.deadLetterQueue.entries()).map(([key, value]) => ({
|
||||
cacheKey: key,
|
||||
...value
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* 从死信队列恢复凭证
|
||||
* @param {string} cacheKey - 缓存键
|
||||
* @returns {boolean} 是否恢复成功
|
||||
*/
|
||||
recoverFromDeadLetter(cacheKey) {
|
||||
const deadEntry = this.deadLetterQueue.get(cacheKey);
|
||||
if (!deadEntry) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 恢复到缓存
|
||||
const entry = deadEntry.entry;
|
||||
entry.retryCount = 0;
|
||||
entry.isDirty = true;
|
||||
entry.lastError = null;
|
||||
entry.firstFailureTime = null;
|
||||
this.credentialCache.set(cacheKey, entry);
|
||||
this.dirtyKeys.add(cacheKey);
|
||||
|
||||
// 从死信队列移除
|
||||
this.deadLetterQueue.delete(cacheKey);
|
||||
|
||||
console.log(`[CredentialCache] Recovered credential from dead letter queue: ${cacheKey}`);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 清除所有缓存（用于测试）
|
||||
*/
|
||||
clear() {
|
||||
this.credentialCache.clear();
|
||||
this.lockChains.clear();
|
||||
this.dirtyKeys.clear();
|
||||
this.deadLetterQueue.clear();
|
||||
this.isInitialized = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新加载指定凭证（从文件）
|
||||
* @param {string} providerType - 提供商类型
|
||||
* @param {string} uuid - 节点UUID
|
||||
* @returns {Promise<CredentialEntry|null>}
|
||||
*/
|
||||
async reloadCredentials(providerType, uuid) {
|
||||
const cacheKey = this._getCacheKey(providerType, uuid);
|
||||
const entry = this.credentialCache.get(cacheKey);
|
||||
|
||||
if (!entry || !entry.credPath) {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
const credentials = await this._loadCredentialsFromFile(entry.credPath);
|
||||
if (credentials) {
|
||||
entry.credentials = credentials;
|
||||
entry.lastModified = Date.now();
|
||||
entry.isDirty = false;
|
||||
entry.retryCount = 0;
|
||||
this.dirtyKeys.delete(cacheKey);
|
||||
return entry;
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`[CredentialCache] Failed to reload ${entry.credPath}: ${error.message}`);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// 导出单例获取函数
|
||||
export function getCredentialCacheManager() {
|
||||
return CredentialCacheManager.getInstance();
|
||||
}
|
||||
Loading…
Reference in a new issue