@@ -37,6 +37,7 @@ import type {
3737SessionsUsageAggregates,
3838SessionsUsageResult,
3939} from "../../shared/usage-types.js";
40+import { runTasksWithConcurrency } from "../../utils/run-with-concurrency.js";
4041import {
4142ErrorCodes,
4243errorShape,
@@ -56,6 +57,7 @@ import type { GatewayRequestHandlers, RespondFn } from "./types.js";
56575758const COST_USAGE_CACHE_TTL_MS = 30_000;
5859const COST_USAGE_CACHE_MAX = 256;
60+const SESSIONS_USAGE_CACHE_READ_CONCURRENCY = 12;
5961const DAY_MS = 24 * 60 * 60 * 1000;
60626163type DateRange = { startMs: number; endMs: number };
@@ -744,7 +746,7 @@ async function loadCostUsageSummaryCached(params: {
744746endMs: params.endMs,
745747config: params.config,
746748requestRefresh: true,
747-refreshMode: "sync-when-empty",
749+refreshMode: "background",
748750})
749751.then((summary) => {
750752setCostUsageCache(cacheKey, {
@@ -1107,43 +1109,75 @@ export const usageHandlers: GatewayRequestHandlers = {
11071109target.missingCostEntries += source.missingCostEntries;
11081110};
110911111110-for (const merged of limitedEntries) {
1111-const agentId = merged.agentId;
1112-let usage: SessionCostSummary | null = null;
1112+const usageByEntryIndex: Array<SessionCostSummary | null> = Array.from(
1113+{ length: limitedEntries.length },
1114+() => null,
1115+);
1116+const usageLoadTasks: Array<
1117+() => Promise<{
1118+entryIndex: number;
1119+cacheStatus: UsageCacheStatus;
1120+summary: SessionCostSummary | null;
1121+}>
1122+> = [];
1123+1124+for (const [entryIndex, merged] of limitedEntries.entries()) {
11131125const includedSessionIds = merged.includedSessionIds ?? [merged.sessionId];
11141126for (const includedSessionId of includedSessionIds) {
11151127const isCurrentSession = includedSessionId === merged.sessionId;
11161128const includedSessionFile = isCurrentSession
11171129 ? merged.sessionFile
11181130 : resolveExistingUsageSessionFile({
11191131sessionId: includedSessionId,
1120- agentId,
1132+agentId: merged.agentId,
11211133});
11221134if (!includedSessionFile) {
11231135continue;
11241136}
1125-const cachedUsage = await loadSessionCostSummaryFromCache({
1126-sessionId: includedSessionId,
1127-sessionEntry: isCurrentSession ? merged.storeEntry : undefined,
1128-sessionFile: includedSessionFile,
1129- config,
1130- agentId,
1131- startMs,
1132- endMs,
1133-refreshMode: "sync-when-empty",
1137+usageLoadTasks.push(async () => {
1138+const cachedUsage = await loadSessionCostSummaryFromCache({
1139+sessionId: includedSessionId,
1140+sessionEntry: isCurrentSession ? merged.storeEntry : undefined,
1141+sessionFile: includedSessionFile,
1142+ config,
1143+agentId: merged.agentId,
1144+ startMs,
1145+ endMs,
1146+refreshMode: "background",
1147+});
1148+return {
1149+ entryIndex,
1150+cacheStatus: cachedUsage.cacheStatus,
1151+summary: cachedUsage.summary,
1152+};
11341153});
1135-cacheStatus = mergeUsageCacheStatus(cacheStatus, cachedUsage.cacheStatus);
1136-const includedUsage = cachedUsage.summary;
1137-if (!includedUsage) {
1138-continue;
1139-}
1140-if (!usage) {
1141-usage = createEmptySessionCostSummary();
1142-usage.sessionId = merged.sessionId;
1143-usage.sessionFile = merged.sessionFile;
1144-}
1145-mergeSessionUsageInto(usage, includedUsage);
11461154}
1155+}
1156+1157+const usageLoadResult = await runTasksWithConcurrency({
1158+tasks: usageLoadTasks,
1159+limit: SESSIONS_USAGE_CACHE_READ_CONCURRENCY,
1160+errorMode: "stop",
1161+});
1162+if (usageLoadResult.hasError) {
1163+throw usageLoadResult.firstError;
1164+}
1165+for (const loaded of usageLoadResult.results) {
1166+cacheStatus = mergeUsageCacheStatus(cacheStatus, loaded.cacheStatus);
1167+if (!loaded.summary) {
1168+continue;
1169+}
1170+const merged = limitedEntries[loaded.entryIndex];
1171+const usage = usageByEntryIndex[loaded.entryIndex] ?? createEmptySessionCostSummary();
1172+usage.sessionId = merged.sessionId;
1173+usage.sessionFile = merged.sessionFile;
1174+mergeSessionUsageInto(usage, loaded.summary);
1175+usageByEntryIndex[loaded.entryIndex] = usage;
1176+}
1177+1178+for (const [entryIndex, merged] of limitedEntries.entries()) {
1179+const agentId = merged.agentId;
1180+const usage = usageByEntryIndex[entryIndex];
1147118111481182if (usage) {
11491183aggregateTotals.input += usage.input;
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。