diff --git a/src/app/[locale]/settings/config/page.tsx b/src/app/[locale]/settings/config/page.tsx
index 37a02e503..af3dcfdb9 100644
--- a/src/app/[locale]/settings/config/page.tsx
+++ b/src/app/[locale]/settings/config/page.tsx
@@ -71,6 +71,7 @@ async function SettingsConfigContent({ locale }: { locale: string }) {
enableHttp2: settings.enableHttp2,
enableOpenaiResponsesWebsocket: settings.enableOpenaiResponsesWebsocket,
enableHighConcurrencyMode: settings.enableHighConcurrencyMode,
+ enableMemoryAdmission: settings.enableMemoryAdmission,
interceptAnthropicWarmupRequests: settings.interceptAnthropicWarmupRequests,
enableThinkingSignatureRectifier: settings.enableThinkingSignatureRectifier,
enableThinkingBudgetRectifier: settings.enableThinkingBudgetRectifier,
diff --git a/src/app/api/admin/system-config/route.ts b/src/app/api/admin/system-config/route.ts
index a5f56127a..153c7415e 100644
--- a/src/app/api/admin/system-config/route.ts
+++ b/src/app/api/admin/system-config/route.ts
@@ -110,6 +110,7 @@ export async function POST(req: Request) {
enableResponseInputRectifier: validated.enableResponseInputRectifier,
streamGateMode: validated.streamGateMode,
affinityIgnoreClientSessionId: validated.affinityIgnoreClientSessionId,
+ enableMemoryAdmission: validated.enableMemoryAdmission,
enableCodexSessionIdCompletion: validated.enableCodexSessionIdCompletion,
enableClaudeMetadataUserIdInjection: validated.enableClaudeMetadataUserIdInjection,
enableResponseFixer: validated.enableResponseFixer,
diff --git a/src/app/v1/_lib/proxy-handler.ts b/src/app/v1/_lib/proxy-handler.ts
index c87f19042..fc4cec3a4 100644
--- a/src/app/v1/_lib/proxy-handler.ts
+++ b/src/app/v1/_lib/proxy-handler.ts
@@ -1,9 +1,9 @@
import type { Context } from "hono";
import { isRawPassthroughEndpointPolicy } from "@/app/v1/_lib/proxy/endpoint-policy";
import { findSafeDatabaseError } from "@/drizzle/admitted-client";
-import { getCachedSystemSettings } from "@/lib/config";
+import { getCachedSystemSettings, getCachedSystemSettingsOnlyCache } from "@/lib/config";
import { logger } from "@/lib/logger";
-import { isLocalCapacityError } from "@/lib/memory/governor";
+import { getMemoryGovernor, isLocalCapacityError } from "@/lib/memory/governor";
import { buildLocalCapacityResponse } from "@/lib/memory/http";
import { withRequestMemoryLifetime } from "@/lib/memory/request-lifetime";
import { ProxyStatusTracker } from "@/lib/proxy-status-tracker";
@@ -61,15 +61,13 @@ async function handleOwnedProxyRequest(c: Context): Promise
{
return identity.identity;
};
try {
- session = await ProxySession.fromContext(c);
try {
cachedSystemSettings = await getCachedSystemSettings();
- session.setHighConcurrencyModeEnabled(
- cachedSystemSettings.enableHighConcurrencyMode ?? false
- );
- session.setRawCrossProviderFallbackEnabled(
- cachedSystemSettings.allowNonConversationEndpointProviderFallback ?? true
- );
+ // 正文读取受内存准入约束，必须在读取入站正文之前同步开关。只同步进程缓存中的
+ // 当前设置：缓存失效期间完成的旧查询与读取失败时的默认对象不能改变全进程状态。
+ if (cachedSystemSettings === getCachedSystemSettingsOnlyCache()) {
+ getMemoryGovernor().setEnabled(cachedSystemSettings.enableMemoryAdmission);
+ }
} catch (settingsError) {
const databaseError = findSafeDatabaseError(settingsError);
logger.warn(
@@ -81,9 +79,14 @@ async function handleOwnedProxyRequest(c: Context): Promise {
databaseCode: databaseError?.code,
}
);
- session.setHighConcurrencyModeEnabled(false);
- session.setRawCrossProviderFallbackEnabled(false);
}
+ session = await ProxySession.fromContext(c);
+ session.setHighConcurrencyModeEnabled(cachedSystemSettings?.enableHighConcurrencyMode ?? false);
+ session.setRawCrossProviderFallbackEnabled(
+ cachedSystemSettings
+ ? (cachedSystemSettings.allowNonConversationEndpointProviderFallback ?? true)
+ : false
+ );
// 自动检测请求格式（端点优先，请求体补充）
if (session.originalFormat === "claude") {
diff --git a/src/app/v1/_lib/proxy/session.ts b/src/app/v1/_lib/proxy/session.ts
index e2b2c28bb..3663aa2f5 100644
--- a/src/app/v1/_lib/proxy/session.ts
+++ b/src/app/v1/_lib/proxy/session.ts
@@ -1772,6 +1772,7 @@ async function parseRequestBody(c: Context): Promise {
? getOpenAIImageMultipartSummary(imageRequestMetadata)
: "(multipart image request)";
requestBodyLogNote = "图片 multipart 请求已记录结构化摘要。";
+ loaded.lease.shrinkTo(loaded.retainedBytes);
return {
requestMessage,
@@ -1798,8 +1799,9 @@ async function parseRequestBody(c: Context): Promise {
requestBodyLog = requestBodyText;
requestBodyLogNote = "请求体不是合法 JSON，已记录原始文本。";
}
- // 保留按正文结构估算的工作集，覆盖后续过滤、重试和异步消费者；
- // 不能只按字节数缩至固定倍数，密集小对象的 V8 开销可能更大。
+ // 解析峰值已结束。请求剩余生命周期（含长时间流式响应、过滤、重试与后台消费者）按正文结构
+ // 估算的实际持有量占用额度；密集小对象的 V8 开销计入 structureBytes，不能只按字节数估算。
+ loaded.lease.shrinkTo(loaded.retainedBytes);
return {
requestMessage,
diff --git a/src/app/v1/_lib/proxy/stream-gate/prebuffer-budget.ts b/src/app/v1/_lib/proxy/stream-gate/prebuffer-budget.ts
index 8a111849c..16fc64e29 100644
--- a/src/app/v1/_lib/proxy/stream-gate/prebuffer-budget.ts
+++ b/src/app/v1/_lib/proxy/stream-gate/prebuffer-budget.ts
@@ -55,7 +55,10 @@ export class StreamGatePrebufferBudget {
constructor(
private readonly resolveLimit: () => number,
private readonly governor?: MemoryGovernor
- ) {}
+ ) {
+ // 内存准入关闭时子限额变为不限，已在队列中的请求需要立即放行。
+ governor?.onEnabledChange(() => this.drainWaiters());
+ }
async acquire(reservedBytes: number, signal?: AbortSignal): Promise {
const started = performance.now();
@@ -274,12 +277,14 @@ export function getStreamGatePrebufferBudget(): StreamGatePrebufferBudget {
const globalState = globalThis as typeof globalThis & {
[STREAM_GATE_PREBUFFER_BUDGET_SYMBOL]?: StreamGatePrebufferBudget;
};
+ const governor = getMemoryGovernor();
+ // 门控子限额属于内存准入；准入关闭时不限额，排队者在下一次归还时全部放行。
globalState[STREAM_GATE_PREBUFFER_BUDGET_SYMBOL] ??= new StreamGatePrebufferBudget(
() =>
- process.env.STREAM_GATE_GLOBAL_PREBUFFER_BYTE_CAP
+ governor.enabled && process.env.STREAM_GATE_GLOBAL_PREBUFFER_BYTE_CAP
? resolveStreamGateGlobalPrebufferByteCap()
: Number.MAX_SAFE_INTEGER,
- getMemoryGovernor()
+ governor
);
return globalState[STREAM_GATE_PREBUFFER_BUDGET_SYMBOL];
}
diff --git a/src/drizzle/schema.ts b/src/drizzle/schema.ts
index 4ea75e08b..ba7d80909 100644
--- a/src/drizzle/schema.ts
+++ b/src/drizzle/schema.ts
@@ -1087,6 +1087,10 @@ export const systemSettings = pgTable('system_settings', {
// F3b 最长前缀匹配缓存模拟开关覆写（null = 跟随环境变量 ENABLE_CACHE_EFFECTIVENESS）
cacheEffectivenessEnabled: boolean('cache_effectiveness_enabled'),
+ // 内存准入（默认关闭）
+ // 开启后：请求正文与流式门控前缀按可用内存预算准入，超出时落盘或返回本地 429
+ enableMemoryAdmission: boolean('enable_memory_admission').notNull().default(false),
+
createdAt: timestamp('created_at', { withTimezone: true }).defaultNow(),
updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow(),
}, (table) => ({
diff --git a/src/lib/api-client/v1/openapi-types.gen.ts b/src/lib/api-client/v1/openapi-types.gen.ts
index 037675e5e..76e147c1d 100644
--- a/src/lib/api-client/v1/openapi-types.gen.ts
+++ b/src/lib/api-client/v1/openapi-types.gen.ts
@@ -12744,6 +12744,8 @@ export interface operations {
streamGateMode: "off" | "shadow" | "enforce";
/** @description Whether fingerprintable requests force longest-prefix affinity for provider stickiness, skipping client session id binding. */
affinityIgnoreClientSessionId: boolean;
+ /** @description Whether local memory admission is enabled. When enabled, request bodies and stream gate prefixes are admitted against the available memory budget, spill to disk, and return a local 429 when capacity runs out. When disabled, they stay in memory without queuing or local 429. */
+ enableMemoryAdmission: boolean;
/** @description Request replay (response caching and upstream connection reuse) override. Null follows the ENABLE_REQUEST_REPLAY environment variable. */
replayEnabled: boolean | null;
/** @description Replay completed payload reuse window in minutes. */
@@ -13034,6 +13036,8 @@ export interface operations {
streamGateMode?: "off" | "shadow" | "enforce";
/** @description Whether fingerprintable requests force longest-prefix affinity for provider stickiness, skipping client session id binding. */
affinityIgnoreClientSessionId?: boolean;
+ /** @description Whether local memory admission is enabled. When enabled, request bodies and stream gate prefixes are admitted against the available memory budget, spill to disk, and return a local 429 when capacity runs out. When disabled, they stay in memory without queuing or local 429. */
+ enableMemoryAdmission?: boolean;
/** @description Request replay (response caching and upstream connection reuse) override. Null follows the ENABLE_REQUEST_REPLAY environment variable. */
replayEnabled?: boolean | null;
/** @description Replay completed payload reuse window in minutes. */
@@ -13199,6 +13203,8 @@ export interface operations {
streamGateMode: "off" | "shadow" | "enforce";
/** @description Whether fingerprintable requests force longest-prefix affinity for provider stickiness, skipping client session id binding. */
affinityIgnoreClientSessionId: boolean;
+ /** @description Whether local memory admission is enabled. When enabled, request bodies and stream gate prefixes are admitted against the available memory budget, spill to disk, and return a local 429 when capacity runs out. When disabled, they stay in memory without queuing or local 429. */
+ enableMemoryAdmission: boolean;
/** @description Request replay (response caching and upstream connection reuse) override. Null follows the ENABLE_REQUEST_REPLAY environment variable. */
replayEnabled: boolean | null;
/** @description Replay completed payload reuse window in minutes. */
diff --git a/src/lib/api/v1/schemas/system-config.ts b/src/lib/api/v1/schemas/system-config.ts
index 2b61000d3..4ced04e5f 100644
--- a/src/lib/api/v1/schemas/system-config.ts
+++ b/src/lib/api/v1/schemas/system-config.ts
@@ -234,6 +234,11 @@ export const SystemSettingsSchema = z
.describe(
"Whether fingerprintable requests force longest-prefix affinity for provider stickiness, skipping client session id binding."
),
+ enableMemoryAdmission: z
+ .boolean()
+ .describe(
+ "Whether local memory admission is enabled. When enabled, request bodies and stream gate prefixes are admitted against the available memory budget, spill to disk, and return a local 429 when capacity runs out. When disabled, they stay in memory without queuing or local 429."
+ ),
replayEnabled: z
.boolean()
.nullable()
diff --git a/src/lib/body-store/allocation-estimate.ts b/src/lib/body-store/allocation-estimate.ts
index 8f60b93e6..74d638965 100644
--- a/src/lib/body-store/allocation-estimate.ts
+++ b/src/lib/body-store/allocation-estimate.ts
@@ -23,4 +23,13 @@ export class AllocationEstimate {
// 明文字节、UTF-16 解析临时值、对象中的字符串及出站序列化工作集。
return 8192 + this.bytes * 8 + this.structureBytes;
}
+
+ /**
+ * 解析完成后请求在整个生命周期内持续持有的工作集：原始字节、对象中的字符串
+ * （UTF-16 最多两倍）、对象结构与一份出站序列化副本。解析临时值已可回收，
+ * 长时间流式响应期间不能继续按解析峰值占用额度。
+ */
+ get retainedBytes(): number {
+ return 8192 + this.bytes * 4 + this.structureBytes;
+ }
}
diff --git a/src/lib/body-store/byte-store.ts b/src/lib/body-store/byte-store.ts
index 32692f4b8..1666081c0 100644
--- a/src/lib/body-store/byte-store.ts
+++ b/src/lib/body-store/byte-store.ts
@@ -2,7 +2,7 @@ import { type FileHandle, mkdir, mkdtemp, open, rm, statfs } from "node:fs/promi
import path from "node:path";
import { BufferedByteChunks } from "@/app/v1/_lib/proxy/buffered-byte-chunks";
import { logger } from "@/lib/logger";
-import { LocalCapacityError, type MemoryLease } from "@/lib/memory/governor";
+import { getMemoryGovernor, LocalCapacityError, type MemoryLease } from "@/lib/memory/governor";
import { getSpoolBudget, getSpoolRoot, spoolPrefix } from "../../../server-lib/spool-directory";
export const STORE_SCRATCH_BYTES = 128 * 1024;
@@ -93,11 +93,10 @@ export class ByteStore {
private async appendInternal(chunk: Uint8Array): Promise {
if (this.disposed) throw new Error("Body store disposed");
const capacity = Math.ceil((this.byteLength + chunk.byteLength) / BLOCK_BYTES) * BLOCK_BYTES;
- if (
- !this.file &&
- capacity <= (this.options.hotBytes ?? HOT_BYTES) &&
- (await this.tryGrow(this.scratchBytes + capacity))
- ) {
+ // 内存准入关闭时正文全部留在内存，不落盘。
+ const hotBytes =
+ this.options.hotBytes ?? (getMemoryGovernor().enabled ? HOT_BYTES : Number.POSITIVE_INFINITY);
+ if (!this.file && capacity <= hotBytes && (await this.tryGrow(this.scratchBytes + capacity))) {
this.memory.append(chunk);
this.byteLength += chunk.byteLength;
return;
diff --git a/src/lib/body-store/request-body-store.ts b/src/lib/body-store/request-body-store.ts
index 02676b608..eacdaa6db 100644
--- a/src/lib/body-store/request-body-store.ts
+++ b/src/lib/body-store/request-body-store.ts
@@ -36,6 +36,8 @@ export async function loadRequestBody(
originalByteLength: number;
encoding: string | null;
lease: MemoryLease;
+ /** 调用方完成解析后应把 lease 收缩到该值。 */
+ retainedBytes: number;
}> {
const governor = getMemoryGovernor();
const started = performance.now();
@@ -193,7 +195,10 @@ export async function loadRequestBody(
source.restoreMemoryAfterAdmission();
}
const heap = getHeapStatistics();
- if (estimate.capacityBytes > (heap.heap_size_limit - heap.used_heap_size) * 0.5) {
+ if (
+ governor.enabled &&
+ estimate.capacityBytes > (heap.heap_size_limit - heap.used_heap_size) * 0.5
+ ) {
throw new LocalCapacityError();
}
const buffer = await source.arrayBuffer();
@@ -201,7 +206,7 @@ export async function loadRequestBody(
governor.observe("body_materialize", performance.now() - materializeStarted, source.byteLength);
const lease = materializedLease;
materializedLease = null;
- return { buffer, encoding, originalByteLength, lease };
+ return { buffer, encoding, originalByteLength, lease, retainedBytes: estimate.retainedBytes };
} finally {
await Promise.allSettled([
raw.dispose(() => rawLease.release()),
diff --git a/src/lib/config/system-settings-cache.ts b/src/lib/config/system-settings-cache.ts
index 15d814c03..c28d5f7fb 100644
--- a/src/lib/config/system-settings-cache.ts
+++ b/src/lib/config/system-settings-cache.ts
@@ -128,6 +128,7 @@ export const DEFAULT_SETTINGS: Pick<
| "racingTotalTimeoutMs"
| "stickyTimeoutCooldownMs"
| "legacyHedgeMaxInFlight"
+ | "enableMemoryAdmission"
> = {
enableHttp2: false,
enableOpenaiResponsesWebsocket: true,
@@ -169,6 +170,7 @@ export const DEFAULT_SETTINGS: Pick<
racingTotalTimeoutMs: 60_000,
stickyTimeoutCooldownMs: 300_000,
legacyHedgeMaxInFlight: 2,
+ enableMemoryAdmission: false,
};
/**
@@ -263,6 +265,7 @@ export async function getCachedSystemSettings(): Promise {
replayEnabled: null,
replayCacheTtlMinutes: DEFAULT_SETTINGS.replayCacheTtlMinutes,
cacheEffectivenessEnabled: null,
+ enableMemoryAdmission: DEFAULT_SETTINGS.enableMemoryAdmission,
discoveryEnabled: DEFAULT_SETTINGS.discoveryEnabled,
discoveryConcurrency: DEFAULT_SETTINGS.discoveryConcurrency,
maxDiscoveryRounds: DEFAULT_SETTINGS.maxDiscoveryRounds,
diff --git a/src/lib/memory/governor.ts b/src/lib/memory/governor.ts
index 5561b3e1d..7def0b41d 100644
--- a/src/lib/memory/governor.ts
+++ b/src/lib/memory/governor.ts
@@ -25,6 +25,10 @@ export interface MemoryLeaseLedger {
byTag: Record;
}
export interface MemoryGovernor {
+ /** 系统设置「内存准入」开关；关闭时租约与增长总是成功。 */
+ readonly enabled: boolean;
+ setEnabled(enabled: boolean): void;
+ onEnabledChange(listener: (enabled: boolean) => void): () => void;
observe(
stage: "admission" | "body_read" | "body_decode" | "body_materialize" | "gate",
milliseconds: number,
@@ -38,6 +42,7 @@ export interface MemoryGovernor {
): Promise;
tryLease(bytes: number, tag?: MemoryLeaseTag): MemoryLease | null;
snapshot(): {
+ enabled: boolean;
usedBytes: number;
limitBytes: number;
waiting: number;
diff --git a/src/lib/validation/schemas.ts b/src/lib/validation/schemas.ts
index 701b36e40..1bdc236f3 100644
--- a/src/lib/validation/schemas.ts
+++ b/src/lib/validation/schemas.ts
@@ -1171,6 +1171,8 @@ export const UpdateSystemSettingsSchema = z
.optional(),
// 忽略客户端 Session ID（可选）
affinityIgnoreClientSessionId: z.boolean().optional(),
+ // 内存准入（可选）
+ enableMemoryAdmission: z.boolean().optional(),
// F2 Replay 响应缓存与复用（可选；null = 跟随环境变量）
replayEnabled: z.boolean().nullable().optional(),
// F2 Replay 完成 payload 可重放窗口(分钟)
diff --git a/src/repository/_shared/transformers.ts b/src/repository/_shared/transformers.ts
index 8c3567140..2bf75328d 100644
--- a/src/repository/_shared/transformers.ts
+++ b/src/repository/_shared/transformers.ts
@@ -345,6 +345,7 @@ export function toSystemSettings(dbSettings: any): SystemSettings {
replayEnabled: dbSettings?.replayEnabled ?? null,
replayCacheTtlMinutes: normalizedReplayCacheTtlMinutes,
cacheEffectivenessEnabled: dbSettings?.cacheEffectivenessEnabled ?? null,
+ enableMemoryAdmission: dbSettings?.enableMemoryAdmission ?? false,
createdAt: dbSettings?.createdAt ? new Date(dbSettings.createdAt) : new Date(),
updatedAt: dbSettings?.updatedAt ? new Date(dbSettings.updatedAt) : new Date(),
};
diff --git a/src/repository/system-config.ts b/src/repository/system-config.ts
index 22e7a9f98..a400c3cea 100644
--- a/src/repository/system-config.ts
+++ b/src/repository/system-config.ts
@@ -207,6 +207,7 @@ function createFallbackSettings(): SystemSettings {
replayEnabled: null,
replayCacheTtlMinutes: REPLAY_CACHE_TTL_MINUTES_DEFAULT,
cacheEffectivenessEnabled: null,
+ enableMemoryAdmission: false,
createdAt: now,
updatedAt: now,
};
@@ -285,6 +286,12 @@ const RECENT_COLUMN_LADDER: ReadonlyArray<{
// 本层更新失败（仍有列缺失）时记录的告警
updateWarn: string;
}> = [
+ {
+ key: "enableMemoryAdmission",
+ column: systemSettings.enableMemoryAdmission,
+ selectWarn: "system_settings 缺少 enableMemoryAdmission，回退到上一代字段集。",
+ updateWarn: "system_settings 缺少 enableMemoryAdmission，继续降级更新。",
+ },
{
key: "legacyHedgeMaxInFlight",
column: systemSettings.legacyHedgeMaxInFlight,
@@ -927,6 +934,11 @@ export async function updateSystemSettings(
updates.cacheEffectivenessEnabled = payload.cacheEffectivenessEnabled;
}
+ // 内存准入开关（如果提供）
+ if (payload.enableMemoryAdmission !== undefined) {
+ updates.enableMemoryAdmission = payload.enableMemoryAdmission;
+ }
+
let updated;
try {
[updated] = await executor
diff --git a/src/types/system-config.ts b/src/types/system-config.ts
index 693845799..4aa59806c 100644
--- a/src/types/system-config.ts
+++ b/src/types/system-config.ts
@@ -171,6 +171,10 @@ export interface SystemSettings {
// null = 跟随环境变量 ENABLE_CACHE_EFFECTIVENESS（默认 true）
cacheEffectivenessEnabled: boolean | null;
+ // 内存准入（默认关闭）
+ // 开启后：请求正文与流式门控前缀按可用内存预算准入，超出时落盘或返回本地 429
+ enableMemoryAdmission: boolean;
+
/** Bounded streaming Discovery settings. */
discoveryEnabled: boolean;
discoveryConcurrency: number;
@@ -308,4 +312,7 @@ export interface UpdateSystemSettingsInput {
// F3b 缓存模拟开关（可选；null = 清除覆写跟随环境变量）
cacheEffectivenessEnabled?: boolean | null;
+
+ // 内存准入开关（可选）
+ enableMemoryAdmission?: boolean;
}
diff --git a/tests/integration/proxy-hedge-lifecycle.test.ts b/tests/integration/proxy-hedge-lifecycle.test.ts
index 84c945a2d..ec6743109 100644
--- a/tests/integration/proxy-hedge-lifecycle.test.ts
+++ b/tests/integration/proxy-hedge-lifecycle.test.ts
@@ -96,6 +96,7 @@ vi.mock("@/lib/memory/governor", async (importOriginal) => {
limit: 64 * 1024 ** 2,
remote: false,
monitor: false,
+ enabled: true,
});
return state.bodyGovernor;
},
@@ -753,7 +754,12 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p
const client = new AbortController();
const now = vi.spyOn(Date, "now");
const neutralPrefixConsumption = watchNeutralResponsesPrefixConsumption();
- const governor = new MemoryGovernor({ limit: 8 * 1024 ** 2, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 8 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 8 * 1024 ** 2, governor);
const admission = vi
.spyOn(getStreamGatePrebufferBudget(), "acquire")
@@ -837,7 +843,12 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p
async ({ sticky, waitMs, cancel }) => {
const [first, second] = await Promise.all([startUpstream(), startUpstream()]);
const client = new AbortController();
- const governor = new MemoryGovernor({ limit: 1024 ** 2, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 1024 ** 2, governor);
const occupied = await budget.acquire(1024 ** 2);
const admission = vi
@@ -916,7 +927,12 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p
it("pauses Sticky and race deadlines during compatibility response admission", async () => {
const upstream = await startUpstream("application/json");
const client = new AbortController();
- const governor = new MemoryGovernor({ limit: 1024 ** 2, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 1024 ** 2, governor);
const entered = Promise.withResolvers();
let occupied: Awaited> | undefined;
@@ -1000,7 +1016,12 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p
it("Discovery 解析容量不足返回本地 429，取消所有候选且不惩罚供应商", async () => {
const [first, second] = await Promise.all([startUpstream(), startUpstream()]);
const client = new AbortController();
- const governor = new MemoryGovernor({ limit: 256 * 1024, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 256 * 1024,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 256 * 1024, governor);
const admission = vi
.spyOn(getStreamGatePrebufferBudget(), "acquire")
diff --git a/tests/unit/actions/system-config-memory-admission-settings.test.ts b/tests/unit/actions/system-config-memory-admission-settings.test.ts
new file mode 100644
index 000000000..3c541f304
--- /dev/null
+++ b/tests/unit/actions/system-config-memory-admission-settings.test.ts
@@ -0,0 +1,50 @@
+import { describe, expect, test, vi } from "vitest";
+
+vi.mock("server-only", () => ({}));
+
+describe("enableMemoryAdmission system setting", () => {
+ test("defaults to disabled in the DB-row transformer and keeps an explicit value", async () => {
+ const { toSystemSettings } = await import("@/repository/_shared/transformers");
+
+ expect(toSystemSettings(undefined).enableMemoryAdmission).toBe(false);
+ expect(toSystemSettings({ id: 1, siteTitle: "CC Hub" }).enableMemoryAdmission).toBe(false);
+ expect(toSystemSettings({ id: 1, enableMemoryAdmission: true }).enableMemoryAdmission).toBe(
+ true
+ );
+ });
+
+ test("is accepted by the settings update validation schema", async () => {
+ const { UpdateSystemSettingsSchema } = await import("@/lib/validation/schemas");
+
+ expect(UpdateSystemSettingsSchema.parse({ enableMemoryAdmission: true })).toMatchObject({
+ enableMemoryAdmission: true,
+ });
+ expect(UpdateSystemSettingsSchema.parse({}).enableMemoryAdmission).toBeUndefined();
+ expect(() => UpdateSystemSettingsSchema.parse({ enableMemoryAdmission: "yes" })).toThrow();
+ });
+
+ test("is exposed by the v1 system settings response and update schemas", async () => {
+ const { SystemSettingsSchema, SystemSettingsUpdateSchema } = await import(
+ "@/lib/api/v1/schemas/system-config"
+ );
+
+ expect(Object.keys(SystemSettingsSchema.shape)).toContain("enableMemoryAdmission");
+ expect(SystemSettingsUpdateSchema.parse({ enableMemoryAdmission: true })).toEqual({
+ enableMemoryAdmission: true,
+ });
+ });
+
+ test("defaults to disabled in the settings cache fallback", async () => {
+ const { DEFAULT_SETTINGS } = await import("@/lib/config/system-settings-cache");
+
+ expect(DEFAULT_SETTINGS.enableMemoryAdmission).toBe(false);
+ });
+
+ test("is stored as a non-null column that defaults to false", async () => {
+ const { systemSettings } = await import("@/drizzle/schema");
+
+ expect(systemSettings.enableMemoryAdmission.name).toBe("enable_memory_admission");
+ expect(systemSettings.enableMemoryAdmission.notNull).toBe(true);
+ expect(systemSettings.enableMemoryAdmission.default).toBe(false);
+ });
+});
diff --git a/tests/unit/actions/system-config-save.test.ts b/tests/unit/actions/system-config-save.test.ts
index df00e73db..28a46a49e 100644
--- a/tests/unit/actions/system-config-save.test.ts
+++ b/tests/unit/actions/system-config-save.test.ts
@@ -420,4 +420,18 @@ describe("saveSystemSettings", () => {
})
);
});
+
+ it("should pass enableMemoryAdmission through validation and save, then invalidate the cache", async () => {
+ const result = await saveSystemSettings({
+ enableMemoryAdmission: true,
+ });
+
+ expect(result.ok).toBe(true);
+ expect(updateSystemSettingsMock).toHaveBeenCalledWith(
+ expect.objectContaining({
+ enableMemoryAdmission: true,
+ })
+ );
+ expect(invalidateSystemSettingsCacheMock).toHaveBeenCalledTimes(1);
+ });
});
diff --git a/tests/unit/proxy/memory-aware-body-store.test.ts b/tests/unit/proxy/memory-aware-body-store.test.ts
index 5392826f9..4f470eea0 100644
--- a/tests/unit/proxy/memory-aware-body-store.test.ts
+++ b/tests/unit/proxy/memory-aware-body-store.test.ts
@@ -18,11 +18,17 @@ let testDirectory: string;
beforeEach(async () => {
// Exercise the real admission implementation without depending on the host's
// instantaneous free RAM when the parallel test worker first loads it.
- const governor = new MemoryGovernor({ limit: 64 * 1024 ** 2, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 64 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
vi.spyOn(getMemoryGovernor(), "acquire").mockImplementation((...args) =>
governor.acquire(...args)
);
vi.spyOn(getMemoryGovernor(), "tryLease").mockImplementation((bytes) => governor.tryLease(bytes));
+ getMemoryGovernor().setEnabled(true);
// 使用工作区所在磁盘，避免 Linux 的 /tmp 挂载为 tmpfs。
const root = path.join(process.cwd(), "tmp");
await mkdir(root, { recursive: true });
@@ -31,6 +37,7 @@ beforeEach(async () => {
});
afterEach(async () => {
vi.unstubAllEnvs();
+ getMemoryGovernor().setEnabled(false);
await rm(testDirectory, { recursive: true, force: true });
});
describe("内存/磁盘共享正文", () => {
@@ -40,6 +47,7 @@ describe("内存/磁盘共享正文", () => {
limit: STORE_SCRATCH_BYTES,
monitor: false,
remote: false,
+ enabled: true,
});
const lease = await governor.acquire(STORE_SCRATCH_BYTES);
const store = new ByteStore(lease, { directory, ioTimeoutMs: 5000 });
@@ -67,6 +75,7 @@ describe("内存/磁盘共享正文", () => {
limit: STORE_SCRATCH_BYTES,
monitor: false,
remote: false,
+ enabled: true,
});
const lease = await governor.acquire(STORE_SCRATCH_BYTES);
const store = new ByteStore(lease, { directory, maxDiskBytes: 32 });
@@ -81,7 +90,12 @@ describe("内存/磁盘共享正文", () => {
}
});
it("小正文保持内存并持有自有块，避免小视图挂住大 backing buffer", async () => {
- const governor = new MemoryGovernor({ limit: 1024 * 1024, monitor: false, remote: false });
+ const governor = new MemoryGovernor({
+ limit: 1024 * 1024,
+ monitor: false,
+ remote: false,
+ enabled: true,
+ });
const lease = await governor.acquire(STORE_SCRATCH_BYTES);
const store = new ByteStore(lease);
const parent = new Uint8Array(16 * 1024 * 1024);
diff --git a/tests/unit/proxy/memory-aware-contention.test.ts b/tests/unit/proxy/memory-aware-contention.test.ts
index af0a994c6..4ecc65665 100644
--- a/tests/unit/proxy/memory-aware-contention.test.ts
+++ b/tests/unit/proxy/memory-aware-contention.test.ts
@@ -11,7 +11,10 @@ import { getMemoryGovernor } from "@/lib/memory/governor";
import { createMemoryCoordinator } from "../../../server-lib/memory-coordinator";
import { MemoryGovernor } from "../../../server-lib/memory-governor";
-afterEach(() => vi.unstubAllEnvs());
+afterEach(() => {
+ vi.unstubAllEnvs();
+ getMemoryGovernor().setEnabled(false);
+});
describe("request capacity contention", () => {
it("negotiates remote credits before rejecting a valid Discovery prefix", async () => {
@@ -32,7 +35,12 @@ describe("request capacity contention", () => {
});
coordinator.attach(worker);
coordinator.resetBaseline();
- const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+ const governor = new MemoryGovernor({
+ processRef: child,
+ remote: true,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => Number.MAX_SAFE_INTEGER, governor);
const prefix = new DiscoveryPrebuffer();
try {
@@ -66,7 +74,9 @@ describe("request capacity contention", () => {
limit: compressed ? 384 * 1024 : 1024 ** 2,
remote: false,
monitor: false,
+ enabled: true,
});
+ getMemoryGovernor().setEnabled(true);
vi.spyOn(getMemoryGovernor(), "acquire").mockImplementation((...args) =>
governor.acquire(...args)
);
diff --git a/tests/unit/proxy/memory-aware-discovery.test.ts b/tests/unit/proxy/memory-aware-discovery.test.ts
index fa81f114b..a14a8e962 100644
--- a/tests/unit/proxy/memory-aware-discovery.test.ts
+++ b/tests/unit/proxy/memory-aware-discovery.test.ts
@@ -14,7 +14,12 @@ const frame = (payload: object) => encoder.encode(`data: ${JSON.stringify(payloa
describe("Discovery 前缀与解析容量", () => {
it("中性帧实际增长、EOF 不提前归还，赢家回放后释放", async () => {
- const governor = new MemoryGovernor({ limit: 8 * 1024 ** 2, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 8 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 8 * 1024 ** 2, governor);
const neutral = frame({
type: "response.created",
@@ -59,7 +64,12 @@ describe("Discovery 前缀与解析容量", () => {
});
it("增长不足在解析与复制前返回本地 429，取消后两级额度归零", async () => {
- const governor = new MemoryGovernor({ limit: 256 * 1024, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 256 * 1024,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 256 * 1024, governor);
const buffer = new DiscoveryPrebuffer();
buffer.attachLease(await budget.acquire(128 * 1024));
diff --git a/tests/unit/proxy/memory-aware-disk-timeout.test.ts b/tests/unit/proxy/memory-aware-disk-timeout.test.ts
index 9e19c43fb..192925d5d 100644
--- a/tests/unit/proxy/memory-aware-disk-timeout.test.ts
+++ b/tests/unit/proxy/memory-aware-disk-timeout.test.ts
@@ -1,8 +1,8 @@
// @vitest-environment node
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { ByteStore, STORE_SCRATCH_BYTES } from "@/lib/body-store/byte-store";
import { runStreamContentGate } from "@/app/v1/_lib/proxy/stream-gate/stream-content-gate";
-import { LocalCapacityError } from "@/lib/memory/governor";
+import { getMemoryGovernor, LocalCapacityError } from "@/lib/memory/governor";
import { logger } from "@/lib/logger";
import { MemoryGovernor } from "../../../server-lib/memory-governor";
import { getSpoolBudget } from "../../../server-lib/spool-directory";
@@ -19,9 +19,11 @@ const disk = vi.hoisted(() => ({
}));
vi.mock("node:fs/promises", () => disk);
vi.mock("@/lib/logger", () => ({ logger: { warn: vi.fn() } }));
+beforeEach(() => getMemoryGovernor().setEnabled(true));
afterEach(() => {
vi.useRealTimers();
vi.unstubAllEnvs();
+ getMemoryGovernor().setEnabled(false);
});
describe("磁盘故障与取消", () => {
@@ -33,6 +35,7 @@ describe("磁盘故障与取消", () => {
limit: STORE_SCRATCH_BYTES,
remote: false,
monitor: false,
+ enabled: true,
});
const lease = governor.tryLease(STORE_SCRATCH_BYTES)!;
return { governor, lease };
@@ -90,6 +93,7 @@ describe("磁盘故障与取消", () => {
limit: STORE_SCRATCH_BYTES * 2 + 4096,
remote: false,
monitor: false,
+ enabled: true,
});
const lease = governor.tryLease(STORE_SCRATCH_BYTES)!;
const store = new ByteStore(lease);
diff --git a/tests/unit/proxy/memory-aware-lifetime.test.ts b/tests/unit/proxy/memory-aware-lifetime.test.ts
index cf4c4bfe6..a109353b2 100644
--- a/tests/unit/proxy/memory-aware-lifetime.test.ts
+++ b/tests/unit/proxy/memory-aware-lifetime.test.ts
@@ -20,7 +20,12 @@ afterEach(() => {
describe("请求内存确定性回收", () => {
it("小预算连续处理大正文无需 GC，额度在响应 EOF 后归零", async () => {
- const governor = new MemoryGovernor({ limit: 4 * 1024 ** 2, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 4 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
vi.spyOn(getMemoryGovernor(), "acquire").mockImplementation((bytes, signal, wait) =>
governor.acquire(bytes, signal, wait)
);
@@ -45,7 +50,12 @@ describe("请求内存确定性回收", () => {
});
it.each(["no-body", "throw", "read-error", "cancel"])("%s 路径只归还一次", async (mode) => {
- const governor = new MemoryGovernor({ limit: 100, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 100,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const lease = governor.tryLease(100)!;
const release = vi.spyOn(lease, "release");
const result = withRequestMemoryLifetime(async () => {
@@ -71,7 +81,12 @@ describe("请求内存确定性回收", () => {
});
it("取消响应后，后台消费者实际结束才释放；多个消费者不重复归还", async () => {
- const governor = new MemoryGovernor({ limit: 100, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 100,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const held = Promise.withResolvers();
let background!: Promise;
let releaseOther!: () => void;
@@ -93,7 +108,12 @@ describe("请求内存确定性回收", () => {
it("任务管理器取消不提前归还仍执行的消费者", async () => {
vi.stubEnv("CI", "true");
- const governor = new MemoryGovernor({ limit: 100, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 100,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const held = Promise.withResolvers();
const response = await withRequestMemoryLifetime(async () => {
attachRequestMemory(governor.tryLease(100)!);
@@ -108,7 +128,7 @@ describe("请求内存确定性回收", () => {
});
it("作用域外使用兜底，已结束作用域的元数据回调不重复释放", async () => {
- const governor = new MemoryGovernor({ limit: 1, remote: false, monitor: false });
+ const governor = new MemoryGovernor({ limit: 1, remote: false, monitor: false, enabled: true });
const lease = governor.tryLease(1)!;
expect(attachRequestMemory(lease)).toBe(false);
retainCurrentRequestMemory()();
@@ -128,7 +148,12 @@ describe("请求内存确定性回收", () => {
});
it("永不结束的后台所有者在宽限到期后被强制归还，并丢弃正文引用", async () => {
vi.useFakeTimers();
- const governor = new MemoryGovernor({ limit: 100, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 100,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const disposed = vi.fn();
const forcedBefore = getRequestMemoryLifetimeStats().forcedTotal;
const response = await withRequestMemoryLifetime(async () => {
@@ -156,7 +181,12 @@ describe("请求内存确定性回收", () => {
it("仍在推进的后台消费者刷新宽限；响应未结束时不计时", async () => {
vi.useFakeTimers();
- const governor = new MemoryGovernor({ limit: 100, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 100,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
let retention!: ReturnType;
const body = new ReadableStream();
const response = await withRequestMemoryLifetime(async () => {
@@ -184,7 +214,12 @@ describe("请求内存确定性回收", () => {
it("正常结束不触发强制路径，也不调用丢弃回调", async () => {
vi.useFakeTimers();
- const governor = new MemoryGovernor({ limit: 100, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 100,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const disposed = vi.fn();
const held = Promise.withResolvers();
const forcedBefore = getRequestMemoryLifetimeStats().forcedTotal;
@@ -204,7 +239,12 @@ describe("请求内存确定性回收", () => {
});
it("作用域结束后再挂载返回 false，由调用方走 GC 兜底而不是抛错", async () => {
- const governor = new MemoryGovernor({ limit: 10, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 10,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
let attachLate!: () => boolean;
const response = await withRequestMemoryLifetime(async () => {
const lease = governor.tryLease(10)!;
diff --git a/tests/unit/proxy/memory-aware-retained-lease.test.ts b/tests/unit/proxy/memory-aware-retained-lease.test.ts
new file mode 100644
index 000000000..d3b53c09e
--- /dev/null
+++ b/tests/unit/proxy/memory-aware-retained-lease.test.ts
@@ -0,0 +1,159 @@
+import { mkdir, mkdtemp, rm } from "node:fs/promises";
+import path from "node:path";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/drizzle/db", () => ({
+ db: {
+ insert: vi.fn(() => ({
+ values: vi.fn(async () => undefined),
+ })),
+ },
+}));
+
+vi.mock("@/lib/logger", () => ({
+ logger: {
+ debug: vi.fn(),
+ error: vi.fn(),
+ info: vi.fn(),
+ warn: vi.fn(),
+ trace: vi.fn(),
+ fatal: vi.fn(),
+ },
+}));
+
+import { ProxySession } from "@/app/v1/_lib/proxy/session";
+import { AllocationEstimate } from "@/lib/body-store/allocation-estimate";
+import { withRequestMemoryLifetime } from "@/lib/memory/request-lifetime";
+import { MemoryGovernor } from "../../../server-lib/memory-governor";
+
+const GOVERNOR_KEY = Symbol.for("cch.memoryGovernor");
+const globals = globalThis as unknown as Record;
+const previousGovernor = globals[GOVERNOR_KEY];
+let spoolDirectory: string;
+
+beforeEach(async () => {
+ // 使用工作区所在磁盘，避免 Linux 的 /tmp 挂载为 tmpfs。
+ const root = path.join(process.cwd(), "tmp");
+ await mkdir(root, { recursive: true });
+ spoolDirectory = await mkdtemp(path.join(root, "cch-retained-lease-"));
+ vi.stubEnv("CCH_MEMORY_SPILL_DIR", spoolDirectory);
+});
+
+afterEach(async () => {
+ globals[GOVERNOR_KEY] = previousGovernor;
+ vi.unstubAllEnvs();
+ await rm(spoolDirectory, { recursive: true, force: true });
+});
+
+function installGovernor(): MemoryGovernor {
+ const governor = new MemoryGovernor({
+ limit: 256 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
+ globals[GOVERNOR_KEY] = governor;
+ return governor;
+}
+
+function createContext(request: Request) {
+ return {
+ req: {
+ method: request.method,
+ url: request.url,
+ raw: request,
+ header(name?: string) {
+ if (name) return request.headers.get(name) ?? undefined;
+ return Object.fromEntries(request.headers.entries());
+ },
+ },
+ } as never;
+}
+
+function estimateOf(bytes: Uint8Array): AllocationEstimate {
+ const estimate = new AllocationEstimate();
+ estimate.feed(bytes);
+ return estimate;
+}
+
+describe("解析后的请求额度按实际持有量占用", () => {
+ it("持有量估算为原始字节、UTF-16 字符串、结构与出站副本，低于解析峰值", () => {
+ const bytes = new TextEncoder().encode(
+ JSON.stringify({ messages: [{ role: "user", content: "x".repeat(100_000) }] })
+ );
+ const estimate = estimateOf(bytes);
+ const structure = estimate.capacityBytes - 8192 - bytes.length * 8;
+ expect(estimate.retainedBytes).toBe(8192 + bytes.length * 4 + structure);
+ expect(estimate.retainedBytes).toBeLessThan(estimate.capacityBytes);
+ });
+
+ it("JSON 请求解析完成后 body_materialize 额度收缩到持有量，响应结束后归零", async () => {
+ const governor = installGovernor();
+ const text = JSON.stringify({
+ model: "claude-sonnet-4",
+ messages: Array.from({ length: 200 }, (_, index) => ({
+ role: index % 2 ? "assistant" : "user",
+ content: [{ type: "text", text: `turn ${index} `.repeat(200) }],
+ })),
+ });
+ const estimate = estimateOf(new TextEncoder().encode(text));
+
+ const response = await withRequestMemoryLifetime(async () => {
+ const request = new Request("http://localhost/v1/messages", { method: "POST", body: text });
+ const session = await ProxySession.fromContext(createContext(request));
+ expect(session.request.model).toBe("claude-sonnet-4");
+ const ledger = governor.snapshot().leases!;
+ expect(ledger.byTag.body_materialize).toMatchObject({
+ count: 1,
+ bytes: estimate.retainedBytes,
+ });
+ // 解析期间仍按完整峰值预留。
+ expect(governor.snapshot().peakBytes).toBeGreaterThanOrEqual(estimate.capacityBytes);
+ return new Response("OK");
+ });
+ expect(await response.text()).toBe("OK");
+ expect(governor.snapshot().usedBytes).toBe(0);
+ });
+
+ it("非 JSON 正文保留原始文本时同样收缩到持有量", async () => {
+ const governor = installGovernor();
+ const text = "not json ".repeat(50_000);
+ const estimate = estimateOf(new TextEncoder().encode(text));
+
+ const response = await withRequestMemoryLifetime(async () => {
+ const request = new Request("http://localhost/v1/messages", { method: "POST", body: text });
+ const session = await ProxySession.fromContext(createContext(request));
+ expect(session.request.message).toEqual({ raw: text });
+ expect(governor.snapshot().leases!.byTag.body_materialize.bytes).toBe(estimate.retainedBytes);
+ return new Response("OK");
+ });
+ await response.text();
+ expect(governor.snapshot().usedBytes).toBe(0);
+ });
+
+ it("multipart 图片请求解析完成后同样收缩到持有量", async () => {
+ const governor = installGovernor();
+ const formData = new FormData();
+ formData.append("model", "gpt-image-1.5");
+ formData.append("prompt", "draw a cat");
+ formData.append(
+ "image[]",
+ new File([new Uint8Array(200_000).fill(7)], "image.png", { type: "image/png" }),
+ "image.png"
+ );
+ const request = new Request("https://proxy.example.com/v1/images/edits", {
+ method: "POST",
+ body: formData,
+ });
+ const bytes = new Uint8Array(await request.clone().arrayBuffer());
+ const estimate = estimateOf(bytes);
+
+ const response = await withRequestMemoryLifetime(async () => {
+ await ProxySession.fromContext(createContext(request));
+ expect(governor.snapshot().leases!.byTag.body_materialize.bytes).toBe(estimate.retainedBytes);
+ return new Response("OK");
+ });
+ await response.text();
+ expect(governor.snapshot().usedBytes).toBe(0);
+ });
+});
diff --git a/tests/unit/proxy/memory-aware-switch.test.ts b/tests/unit/proxy/memory-aware-switch.test.ts
new file mode 100644
index 000000000..917182e37
--- /dev/null
+++ b/tests/unit/proxy/memory-aware-switch.test.ts
@@ -0,0 +1,220 @@
+import { EventEmitter } from "node:events";
+import { gzipSync } from "node:zlib";
+import { afterAll, afterEach, describe, expect, it, vi } from "vitest";
+import {
+ getStreamGatePrebufferBudget,
+ resolveStreamGateGlobalPrebufferByteCap,
+} from "@/app/v1/_lib/proxy/stream-gate/prebuffer-budget";
+import { ByteStore, STORE_SCRATCH_BYTES } from "@/lib/body-store/byte-store";
+import { loadRequestBody } from "@/lib/body-store/request-body-store";
+import { getMemoryGovernor, LocalCapacityError } from "@/lib/memory/governor";
+import { MemoryGovernor } from "../../../server-lib/memory-governor";
+
+const GATE_CAP = 128 * 1024;
+// getEnvConfig 在进程内只解析一次，门控上限必须在任何模块读取环境变量之前设置。
+const previousGateCap = vi.hoisted(() => {
+ const previous = process.env.STREAM_GATE_GLOBAL_PREBUFFER_BYTE_CAP;
+ process.env.STREAM_GATE_GLOBAL_PREBUFFER_BYTE_CAP = String(128 * 1024);
+ return previous;
+});
+afterAll(() => {
+ if (previousGateCap === undefined) delete process.env.STREAM_GATE_GLOBAL_PREBUFFER_BYTE_CAP;
+ else process.env.STREAM_GATE_GLOBAL_PREBUFFER_BYTE_CAP = previousGateCap;
+});
+
+const GOVERNOR_KEY = Symbol.for("cch.memoryGovernor");
+const BUDGET_KEY = Symbol.for("cch.streamGatePrebufferBudget");
+const globals = globalThis as unknown as Record;
+
+function installGovernor(governor: MemoryGovernor) {
+ const previousGovernor = globals[GOVERNOR_KEY];
+ const previousBudget = globals[BUDGET_KEY];
+ globals[GOVERNOR_KEY] = governor;
+ delete globals[BUDGET_KEY];
+ return () => {
+ globals[GOVERNOR_KEY] = previousGovernor;
+ globals[BUDGET_KEY] = previousBudget;
+ };
+}
+
+afterEach(() => {
+ vi.useRealTimers();
+ vi.unstubAllEnvs();
+});
+
+describe("内存准入开关", () => {
+ it("默认关闭：超出额度的租约与增长立即成功，只记账", async () => {
+ const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false });
+ expect(governor.enabled).toBe(false);
+ expect(governor.snapshot().enabled).toBe(false);
+
+ const lease = governor.tryLease(1024 * 1024, "body_read");
+ expect(lease).not.toBeNull();
+ expect(lease!.tryGrow(4 * 1024 * 1024)).toBe(true);
+ await expect(lease!.tryGrowAsync!(8 * 1024 * 1024)).resolves.toBe(true);
+ const acquired = await governor.acquire(2 * 1024 * 1024);
+ expect(governor.snapshot()).toMatchObject({
+ usedBytes: 10 * 1024 * 1024,
+ waiting: 0,
+ rejected: 0,
+ });
+
+ governor.setEnabled(true);
+ expect(governor.snapshot().enabled).toBe(true);
+ expect(governor.tryLease(1)).toBeNull();
+ expect(lease!.tryGrow(16 * 1024 * 1024)).toBe(false);
+ await expect(governor.acquire(1, undefined, 0)).rejects.toBeInstanceOf(LocalCapacityError);
+
+ lease!.release();
+ acquired.release();
+ expect(governor.snapshot().usedBytes).toBe(0);
+ });
+
+ it("关闭时多进程 worker 不向 primary 申请授权", async () => {
+ const child = Object.assign(new EventEmitter(), {
+ env: {},
+ connected: true,
+ send: vi.fn((_message: unknown, callback?: (error?: Error) => void) => callback?.()),
+ });
+ const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+
+ const lease = await governor.acquire(4 * 1024 * 1024);
+ expect(lease.tryGrow(8 * 1024 * 1024)).toBe(true);
+ await expect(lease.tryGrowAsync!(16 * 1024 * 1024)).resolves.toBe(true);
+ expect(child.send).not.toHaveBeenCalled();
+ expect(governor.snapshot().limitBytes).toBe(0);
+ lease.release();
+ });
+
+ it("排队中的请求在开关关闭后立即放行", async () => {
+ vi.useFakeTimers();
+ const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false, enabled: true });
+ let admitted = false;
+ const waiting = governor.acquire(1024).then((lease) => {
+ admitted = true;
+ return lease;
+ });
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(admitted).toBe(false);
+ expect(governor.snapshot().waiting).toBe(1);
+
+ governor.setEnabled(false);
+ await vi.advanceTimersByTimeAsync(50);
+ expect(admitted).toBe(true);
+ (await waiting).release();
+ expect(governor.snapshot()).toMatchObject({ waiting: 0, rejected: 0, usedBytes: 0 });
+ });
+
+ it("关闭时正文全部留在内存，不创建暂存文件", async () => {
+ const restore = installGovernor(
+ new MemoryGovernor({ limit: 0, remote: false, monitor: false })
+ );
+ vi.stubEnv("CCH_MEMORY_SPILL_DIR", "/nonexistent/cch-spool-must-not-be-used");
+ try {
+ const lease = getMemoryGovernor().tryLease(STORE_SCRATCH_BYTES)!;
+ const store = new ByteStore(lease);
+ const body = new Uint8Array(3 * 1024 * 1024).map((_, index) => index % 251);
+ for (let offset = 0; offset < body.byteLength; offset += 100_000) {
+ await store.append(body.subarray(offset, offset + 100_000));
+ }
+ expect(store.spilled).toBe(false);
+ expect(Buffer.from(await store.arrayBuffer()).equals(Buffer.from(body))).toBe(true);
+ await store.dispose(() => lease.release());
+ } finally {
+ restore();
+ }
+ });
+
+ it("关闭时零额度也能读取并解压大正文", async () => {
+ const restore = installGovernor(
+ new MemoryGovernor({ limit: 0, remote: false, monitor: false })
+ );
+ vi.stubEnv("CCH_MEMORY_SPILL_DIR", "/nonexistent/cch-spool-must-not-be-used");
+ try {
+ const text = JSON.stringify({ model: "claude-test", input: "y".repeat(2 * 1024 * 1024) });
+ const loaded = await loadRequestBody(
+ new Request("http://localhost/v1/messages", {
+ method: "POST",
+ headers: { "content-encoding": "gzip" },
+ body: gzipSync(Buffer.from(text)),
+ })
+ );
+ expect(new TextDecoder().decode(loaded.buffer)).toBe(text);
+ expect(loaded.encoding).toBe("gzip");
+ loaded.lease.release();
+ } finally {
+ restore();
+ }
+ });
+
+ it("开关变化只在状态实际改变时通知，取消订阅后不再通知", () => {
+ const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false });
+ const seen: boolean[] = [];
+ const unsubscribe = governor.onEnabledChange((enabled) => seen.push(enabled));
+ governor.setEnabled(false);
+ governor.setEnabled(true);
+ governor.setEnabled(true);
+ governor.setEnabled(false);
+ unsubscribe();
+ governor.setEnabled(true);
+ expect(seen).toEqual([true, false]);
+ expect(governor.enabled).toBe(true);
+ });
+
+ it("关闭开关时立即放行门控子限额队列中的请求", async () => {
+ vi.useFakeTimers();
+ const governor = new MemoryGovernor({
+ limit: 64 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
+ const restore = installGovernor(governor);
+ try {
+ const budget = getStreamGatePrebufferBudget();
+ const first = await budget.acquire(GATE_CAP);
+ let admitted = false;
+ const queued = budget.acquire(GATE_CAP).then((lease) => {
+ admitted = true;
+ return lease;
+ });
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(admitted).toBe(false);
+ expect(budget.snapshot().waiting).toBe(1);
+
+ governor.setEnabled(false);
+ await vi.advanceTimersByTimeAsync(0);
+ expect(admitted).toBe(true);
+ expect(budget.snapshot().waiting).toBe(0);
+ (await queued).release();
+ first.release();
+ } finally {
+ restore();
+ }
+ });
+
+ it("门控子限额只在开启时生效", async () => {
+ const cap = GATE_CAP;
+ expect(resolveStreamGateGlobalPrebufferByteCap()).toBe(cap);
+ const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false });
+ const restore = installGovernor(governor);
+ try {
+ const budget = getStreamGatePrebufferBudget();
+ expect(budget.snapshot().limit).toBe(Number.MAX_SAFE_INTEGER);
+ const first = await budget.acquire(cap);
+ const second = await budget.acquire(cap);
+ expect(first.tryGrow(32 * cap)).toBe(true);
+ expect(budget.snapshot()).toMatchObject({ waiting: 0, reservedBytes: 33 * cap });
+
+ governor.setEnabled(true);
+ expect(budget.snapshot().limit).toBe(cap);
+ expect(first.tryGrow(64 * cap)).toBe(false);
+
+ first.release();
+ second.release();
+ expect(budget.snapshot().reservedBytes).toBe(0);
+ } finally {
+ restore();
+ }
+ });
+});
diff --git a/tests/unit/proxy/proxy-handler-public-errors.test.ts b/tests/unit/proxy/proxy-handler-public-errors.test.ts
index 5d543823e..ae455c3eb 100644
--- a/tests/unit/proxy/proxy-handler-public-errors.test.ts
+++ b/tests/unit/proxy/proxy-handler-public-errors.test.ts
@@ -10,6 +10,7 @@ type ProxySettingsFixture = {
readonly fakeStreamingWhitelist: FakeStreamingWhitelistEntry[];
readonly passThroughUpstreamErrorMessage: boolean;
readonly verboseProviderError: boolean;
+ readonly enableMemoryAdmission: boolean;
};
const boundary = vi.hoisted(() => ({
@@ -21,6 +22,8 @@ const boundary = vi.hoisted(() => ({
incrementConcurrentCount: vi.fn<(sessionId: string) => Promise>(),
incrementObservedConcurrentCount: vi.fn<(identity: string) => Promise>(),
loadSettings: vi.fn<() => Promise>(),
+ // 进程缓存中的当前设置对象；只有与之相同的读取结果才会同步到内存准入开关。
+ processCache: { current: null as ProxySettingsFixture | null },
recordLocalCapacityRejection: vi.fn(),
recordPreAuthLocalCapacityRejection: vi.fn(),
runGuards: vi.fn<(session: ProxySession) => Promise>(),
@@ -32,13 +35,20 @@ const boundary = vi.hoisted(() => ({
vi.mock("@/lib/config", async (importOriginal) => ({
...(await importOriginal()),
getCachedSystemSettings: boundary.loadSettings,
+ getCachedSystemSettingsOnlyCache: () => boundary.processCache.current,
}));
vi.mock("@/lib/config/system-settings-cache", async (importOriginal) => ({
...(await importOriginal()),
getCachedSystemSettings: boundary.loadSettings,
+ getCachedSystemSettingsOnlyCache: () => boundary.processCache.current,
}));
+function cacheSettings(value: ProxySettingsFixture): void {
+ boundary.processCache.current = value;
+ boundary.loadSettings.mockResolvedValue(value);
+}
+
vi.mock("@/app/v1/_lib/proxy/guard-pipeline", () => ({
GuardPipelineBuilder: {
fromSession: () => ({ run: boundary.runGuards }),
@@ -103,6 +113,7 @@ const settings: ProxySettingsFixture = {
fakeStreamingWhitelist: [],
passThroughUpstreamErrorMessage: false,
verboseProviderError: false,
+ enableMemoryAdmission: false,
};
describe("handleProxyRequest public error behavior", () => {
@@ -122,7 +133,7 @@ describe("handleProxyRequest public error behavior", () => {
boundary.recordPreAuthLocalCapacityRejection.mockReset();
boundary.recordLocalCapacityRejection.mockResolvedValue(true);
boundary.recordPreAuthLocalCapacityRejection.mockResolvedValue(true);
- boundary.loadSettings.mockResolvedValue(settings);
+ cacheSettings(settings);
boundary.getErrorOverride.mockResolvedValue(null);
boundary.incrementConcurrentCount.mockResolvedValue(undefined);
boundary.decrementConcurrentCount.mockResolvedValue(undefined);
@@ -298,6 +309,7 @@ describe("handleProxyRequest public error behavior", () => {
const state = globalThis as unknown as Record;
const previous = state[key];
state[key] = new MemoryGovernor({ limit: 0, remote: false, monitor: false });
+ cacheSettings({ ...settings, enableMemoryAdmission: true });
try {
const request = new Request("http://localhost/v1/messages", {
method: "POST",
@@ -329,4 +341,105 @@ describe("handleProxyRequest public error behavior", () => {
vi.useRealTimers();
}
});
+
+ it("内存准入关闭时请求体读取不排队，零额度也照常进入守卫链", async () => {
+ const key = Symbol.for("cch.memoryGovernor");
+ const state = globalThis as unknown as Record;
+ const previous = state[key];
+ const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false, enabled: true });
+ state[key] = governor;
+ boundary.runGuards.mockResolvedValue(new Response("guarded", { status: 403 }));
+ try {
+ const response = await handleProxyRequest(
+ new Context(
+ new Request("http://localhost/v1/messages", {
+ method: "POST",
+ body: JSON.stringify({ model: "claude-test", messages: [], input: "x".repeat(4096) }),
+ })
+ )
+ );
+ expect(governor.enabled).toBe(false);
+ expect(response.status).toBe(403);
+ expect(boundary.runGuards).toHaveBeenCalledTimes(1);
+ const session = boundary.runGuards.mock.calls[0]?.[0];
+ expect(session?.request.model).toBe("claude-test");
+ expect(boundary.recordPreAuthLocalCapacityRejection).not.toHaveBeenCalled();
+ } finally {
+ state[key] = previous;
+ }
+ });
+
+ it("读取系统设置失败时保持进程当前的内存准入状态", async () => {
+ const key = Symbol.for("cch.memoryGovernor");
+ const state = globalThis as unknown as Record;
+ const previous = state[key];
+ const governor = new MemoryGovernor({
+ limit: 64 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
+ state[key] = governor;
+ boundary.loadSettings.mockRejectedValue(new Error("settings unavailable"));
+ boundary.runGuards.mockResolvedValue(new Response("guarded", { status: 403 }));
+ try {
+ const response = await handleProxyRequest(
+ new Context(
+ new Request("http://localhost/v1/messages", {
+ method: "POST",
+ body: JSON.stringify({ model: "claude-test", messages: [] }),
+ })
+ )
+ );
+ expect(response.status).toBe(403);
+ expect(governor.enabled).toBe(true);
+ const session = boundary.runGuards.mock.calls[0]?.[0];
+ // 设置不可用时 raw 跨供应商回退按既有约定关闭。
+ expect(session?.isRawCrossProviderFallbackEnabled()).toBe(false);
+ } finally {
+ state[key] = previous;
+ }
+ });
+
+ it("缓存失效期间完成的旧查询与默认对象不改变进程的内存准入状态", async () => {
+ const key = Symbol.for("cch.memoryGovernor");
+ const state = globalThis as unknown as Record;
+ const previous = state[key];
+ const governor = new MemoryGovernor({
+ limit: 64 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
+ state[key] = governor;
+ boundary.runGuards.mockImplementation(async () => new Response("guarded", { status: 403 }));
+ const send = () =>
+ handleProxyRequest(
+ new Context(
+ new Request("http://localhost/v1/messages", {
+ method: "POST",
+ body: JSON.stringify({ model: "claude-test", messages: [] }),
+ })
+ )
+ );
+ try {
+ // 失效后缓存为空：读取结果未写入缓存，不能用于全进程开关。
+ boundary.processCache.current = null;
+ boundary.loadSettings.mockResolvedValue({ ...settings, enableMemoryAdmission: false });
+ expect((await send()).status).toBe(403);
+ expect(governor.enabled).toBe(true);
+
+ // 缓存已被更新的设置替换，旧查询结果与之不是同一对象。
+ boundary.processCache.current = { ...settings, enableMemoryAdmission: true };
+ expect((await send()).status).toBe(403);
+ expect(governor.enabled).toBe(true);
+
+ // 当前缓存对象正常同步。
+ cacheSettings({ ...settings, enableMemoryAdmission: false });
+ expect((await send()).status).toBe(403);
+ expect(governor.enabled).toBe(false);
+ } finally {
+ state[key] = previous;
+ }
+ });
});
diff --git a/tests/unit/proxy/routing-trace.test.ts b/tests/unit/proxy/routing-trace.test.ts
index eb2cb6d99..ac91c709f 100644
--- a/tests/unit/proxy/routing-trace.test.ts
+++ b/tests/unit/proxy/routing-trace.test.ts
@@ -327,7 +327,12 @@ describe("ProxySession routing trace recorder", () => {
it.each([false, true])(
"响应 EOF 后保留慢观测刷新和删除的请求额度，删除失败=%s",
async (failed) => {
- const governor = new MemoryGovernor({ limit: 100, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 100,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const write = Promise.withResolvers();
const deletion = Promise.withResolvers();
liveChainMocks.writeLiveRoutingTrace.mockImplementationOnce(() => write.promise);
diff --git a/tests/unit/proxy/stream-gate-content-gate.test.ts b/tests/unit/proxy/stream-gate-content-gate.test.ts
index bac1270e4..4fb6d0d64 100644
--- a/tests/unit/proxy/stream-gate-content-gate.test.ts
+++ b/tests/unit/proxy/stream-gate-content-gate.test.ts
@@ -103,7 +103,12 @@ describe("runStreamContentGate", () => {
it("已提交前缀被丢弃（既未读完也未取消）时，请求作用域结束兜底归还两级额度", async () => {
const reservation = GATE_OPTIONS.prebufferByteCap * 4;
- const governor = new MemoryGovernor({ limit: reservation, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: reservation,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => reservation, governor);
const response = await withRequestMemoryLifetime(async () => {
const committed = await runStreamContentGate(readerFromChunks([TEXT_DELTA]), {
diff --git a/tests/unit/proxy/stream-gate-forwarder-integration.test.ts b/tests/unit/proxy/stream-gate-forwarder-integration.test.ts
index 1bc4cb6d6..18249b4d8 100644
--- a/tests/unit/proxy/stream-gate-forwarder-integration.test.ts
+++ b/tests/unit/proxy/stream-gate-forwarder-integration.test.ts
@@ -597,7 +597,12 @@ describe("F1 stream content gate x ProxyForwarder paths", () => {
});
const session = createSession();
session.setProvider(provider);
- const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 0,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
vi.spyOn(getStreamGatePrebufferBudget(), "acquire").mockImplementation((bytes, signal) =>
governor.acquire(bytes, signal)
);
@@ -637,8 +642,18 @@ describe("F1 stream content gate x ProxyForwarder paths", () => {
const session = createSession();
session.setProvider(first);
mocks.pickRandomProviderWithExclusion.mockResolvedValueOnce(second);
- const free = new MemoryGovernor({ limit: 1024 * 1024, remote: false, monitor: false });
- const occupied = new MemoryGovernor({ limit: 0, remote: false, monitor: false });
+ const free = new MemoryGovernor({
+ limit: 1024 * 1024,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
+ const occupied = new MemoryGovernor({
+ limit: 0,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
vi.spyOn(getStreamGatePrebufferBudget(), "acquire")
.mockImplementationOnce((bytes, signal) => free.acquire(bytes, signal))
.mockImplementation((bytes, signal) => occupied.acquire(bytes, signal));
diff --git a/tests/unit/proxy/stream-gate-ttft-regression.test.ts b/tests/unit/proxy/stream-gate-ttft-regression.test.ts
index acdbccfa3..a76b8e3f5 100644
--- a/tests/unit/proxy/stream-gate-ttft-regression.test.ts
+++ b/tests/unit/proxy/stream-gate-ttft-regression.test.ts
@@ -21,7 +21,12 @@ const options = {
afterEach(() => vi.useRealTimers());
describe("TTFT 固定预占回归", () => {
it("等待门控子限额的请求不占用全局正文额度", async () => {
- const governor = new MemoryGovernor({ limit: 2 * 1024 ** 2, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 2 * 1024 ** 2,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 128 * 1024, governor);
const first = await budget.acquire(128 * 1024);
const controllers = Array.from({ length: 24 }, () => new AbortController());
@@ -42,7 +47,12 @@ describe("TTFT 固定预占回归", () => {
it("本地与全局等待共用 20 秒期限，失败后归还本地子额度", async () => {
vi.useFakeTimers();
- const governor = new MemoryGovernor({ limit: 256, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 256,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const budget = new StreamGatePrebufferBudget(() => 128, governor);
const first = await budget.acquire(128);
const occupied = governor.tryLease(128)!;
diff --git a/tests/unit/repository/system-config-degradation-ladder.test.ts b/tests/unit/repository/system-config-degradation-ladder.test.ts
index 08b0ce198..9ee318bb3 100644
--- a/tests/unit/repository/system-config-degradation-ladder.test.ts
+++ b/tests/unit/repository/system-config-degradation-ladder.test.ts
@@ -7,6 +7,7 @@ import type { UpdateSystemSettingsInput } from "@/types/system-config";
// 近代新增列（最新在前），降级链按引入顺序逐层累计剥离。
const RECENT_COLUMNS = [
+ "enableMemoryAdmission",
"legacyHedgeMaxInFlight",
"replayCacheTtlMinutes",
"cacheEffectivenessEnabled",
@@ -31,6 +32,7 @@ const RECENT_COLUMNS = [
// 全量字段集.
const FULL_COLUMNS = [
+ "enableMemoryAdmission",
"replayCacheTtlMinutes",
"cacheEffectivenessEnabled",
"replayEnabled",
diff --git a/tests/unit/repository/system-config-update-missing-columns.test.ts b/tests/unit/repository/system-config-update-missing-columns.test.ts
index fb0596f82..17664c1f7 100644
--- a/tests/unit/repository/system-config-update-missing-columns.test.ts
+++ b/tests/unit/repository/system-config-update-missing-columns.test.ts
@@ -299,12 +299,13 @@ describe("SystemSettings：数据库缺列时的保存兜底", () => {
vi.setSystemTime(now);
// 第一次 select(fullSelection) 因新列缺失而抛 42703；
- // The new legacy hedge column is the newest rung, so it is stripped before replay columns.
+ // The memory admission and legacy hedge columns are the newest rungs, so they are stripped before replay columns.
const selectMock = vi
.fn()
.mockReturnValueOnce(createRejectedThenableQuery({ code: "42703" }))
.mockReturnValueOnce(createRejectedThenableQuery({ code: "42703" }))
.mockReturnValueOnce(createRejectedThenableQuery({ code: "42703" }))
+ .mockReturnValueOnce(createRejectedThenableQuery({ code: "42703" }))
.mockReturnValueOnce(
createThenableQuery([
{
@@ -337,23 +338,25 @@ describe("SystemSettings：数据库缺列时的保存兜底", () => {
const result = await getSystemSettings();
// 降级读取成功（未抛错），缺失列由 transformer 落默认值。
- expect(selectMock).toHaveBeenCalledTimes(4);
+ expect(selectMock).toHaveBeenCalledTimes(5);
expect(result.siteTitle).toBe("CC Hub");
expect(result.enableHttp2).toBe(true);
expect(result.affinityIgnoreClientSessionId).toBe(true);
expect(result.streamGateMode).toBe("enforce");
-
- const fourthSelection = selectMock.mock.calls[3]?.[0] as Record;
- expect(fourthSelection).not.toHaveProperty("legacyHedgeMaxInFlight");
- expect(fourthSelection).not.toHaveProperty("replayCacheTtlMinutes");
- expect(fourthSelection).not.toHaveProperty("cacheEffectivenessEnabled");
- expect(fourthSelection).toHaveProperty("replayEnabled");
- expect(fourthSelection).toHaveProperty("affinityIgnoreClientSessionId");
- expect(fourthSelection).toHaveProperty("streamGateMode");
- expect(fourthSelection).toHaveProperty("stickyTimeoutCooldownMs");
- expect(fourthSelection).toHaveProperty("racingTotalTimeoutMs");
- expect(fourthSelection).toHaveProperty("enableGeminiFunctionIdRectifier");
- expect(fourthSelection).toHaveProperty("enableThinkingEffortConflictRectifier");
+ expect(result.enableMemoryAdmission).toBe(false);
+
+ const successfulSelection = selectMock.mock.calls[4]?.[0] as Record;
+ expect(successfulSelection).not.toHaveProperty("enableMemoryAdmission");
+ expect(successfulSelection).not.toHaveProperty("legacyHedgeMaxInFlight");
+ expect(successfulSelection).not.toHaveProperty("replayCacheTtlMinutes");
+ expect(successfulSelection).not.toHaveProperty("cacheEffectivenessEnabled");
+ expect(successfulSelection).toHaveProperty("replayEnabled");
+ expect(successfulSelection).toHaveProperty("affinityIgnoreClientSessionId");
+ expect(successfulSelection).toHaveProperty("streamGateMode");
+ expect(successfulSelection).toHaveProperty("stickyTimeoutCooldownMs");
+ expect(successfulSelection).toHaveProperty("racingTotalTimeoutMs");
+ expect(successfulSelection).toHaveProperty("enableGeminiFunctionIdRectifier");
+ expect(successfulSelection).toHaveProperty("enableThinkingEffortConflictRectifier");
vi.useRealTimers();
});
diff --git a/tests/unit/server-memory-governor.test.ts b/tests/unit/server-memory-governor.test.ts
index fadf7d26e..da1184c03 100644
--- a/tests/unit/server-memory-governor.test.ts
+++ b/tests/unit/server-memory-governor.test.ts
@@ -27,7 +27,12 @@ describe("内存租约与本地准入", () => {
},
});
coordinator.attach(worker);
- const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+ const governor = new MemoryGovernor({
+ processRef: child,
+ remote: true,
+ monitor: false,
+ enabled: true,
+ });
let admitted = false;
const waiting = governor.acquire(128 * 1024).then((lease) => {
admitted = true;
@@ -60,7 +65,12 @@ describe("内存租约与本地准入", () => {
});
it("20 秒拒绝，并移除等待者", async () => {
vi.useFakeTimers();
- const governor = new MemoryGovernor({ limit: 1024, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 1024,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const occupied = await governor.acquire(1024);
const waiting = governor.acquire(1);
const assertion = expect(waiting).rejects.toBeInstanceOf(LocalCapacityError);
@@ -72,7 +82,12 @@ describe("内存租约与本地准入", () => {
expect(governor.snapshot().usedBytes).toBe(0);
});
it("扩容失败立即返回，不能持有部分工作集进入等待", async () => {
- const governor = new MemoryGovernor({ limit: 1024, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 1024,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const first = await governor.acquire(512);
const second = await governor.acquire(512);
expect(first.tryGrow(1024)).toBe(false);
@@ -84,7 +99,7 @@ describe("内存租约与本地准入", () => {
expect(first.tryGrow(1)).toBe(false);
});
it("客户端断开立即移除准入等待", async () => {
- const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false });
+ const governor = new MemoryGovernor({ limit: 0, remote: false, monitor: false, enabled: true });
const abort = new AbortController();
const promise = governor.acquire(1, abort.signal);
abort.abort(new Error("closed"));
@@ -132,7 +147,12 @@ describe("内存租约与本地准入", () => {
});
coordinator.resetBaseline();
coordinator.attach(worker);
- const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+ const governor = new MemoryGovernor({
+ processRef: child,
+ remote: true,
+ monitor: false,
+ enabled: true,
+ });
const acquiring = governor.acquire(128 * 1024);
await vi.advanceTimersByTimeAsync(50);
const lease = await acquiring;
@@ -155,7 +175,12 @@ describe("内存租约与本地准入", () => {
throw new Error("closed");
}),
});
- const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+ const governor = new MemoryGovernor({
+ processRef: child,
+ remote: true,
+ monitor: false,
+ enabled: true,
+ });
await governor.requestCredits(100);
expect(governor.snapshot().limitBytes).toBe(0);
child.send.mockImplementation((_message: unknown, callback: (error?: Error) => void) =>
@@ -200,7 +225,12 @@ describe("内存租约与本地准入", () => {
},
});
coordinator.attach(worker);
- const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+ const governor = new MemoryGovernor({
+ processRef: child,
+ remote: true,
+ monitor: false,
+ enabled: true,
+ });
const acquiring = governor.acquire(capacity);
await vi.advanceTimersByTimeAsync(50);
expect(coordinator.snapshot().grantedBytes).toBe(capacity);
@@ -232,6 +262,7 @@ describe("内存租约与本地准入", () => {
const governor = new MemoryGovernor({
remote: false,
monitor: false,
+ enabled: true,
readSnapshot: () => ({
availableRamBytes: ram,
availableSwapBytes: 0,
@@ -280,6 +311,7 @@ describe("内存租约与本地准入", () => {
const governor = new MemoryGovernor({
remote: false,
monitor: false,
+ enabled: true,
readSnapshot: () => ({ availableRamBytes: ram, availableSwapBytes: 0, memoryPressure: 0 }),
});
const ceiling = governor.snapshot().limitBytes;
@@ -314,7 +346,12 @@ describe("内存租约与本地准入", () => {
});
it("租约台账按标签汇总在账量与最长持有时间，归还后移除", async () => {
vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
- const governor = new MemoryGovernor({ limit: 1000, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 1000,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const body = governor.tryLease(300, "body_materialize")!;
vi.setSystemTime(new Date("2026-01-01T00:01:00Z"));
const gate = await governor.acquire(100, undefined, undefined, "gate");
diff --git a/tests/unit/server-memory-growth.test.ts b/tests/unit/server-memory-growth.test.ts
index 1c8384fc9..eb45f467a 100644
--- a/tests/unit/server-memory-growth.test.ts
+++ b/tests/unit/server-memory-growth.test.ts
@@ -23,7 +23,12 @@ function remoteGovernor(capacity = 64 * 1024 ** 2) {
});
coordinator.attach(worker);
coordinator.resetBaseline();
- const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+ const governor = new MemoryGovernor({
+ processRef: child,
+ remote: true,
+ monitor: false,
+ enabled: true,
+ });
return {
governor,
coordinator,
@@ -77,7 +82,12 @@ describe("remote memory growth", () => {
connected: true,
send: vi.fn((_message: unknown, callback?: () => void) => callback?.()),
});
- const governor = new MemoryGovernor({ processRef: child, remote: true, monitor: false });
+ const governor = new MemoryGovernor({
+ processRef: child,
+ remote: true,
+ monitor: false,
+ enabled: true,
+ });
governor.credits = 1024 ** 2;
const lease = governor.tryLease(128 * 1024)!;
const controller = new AbortController();
@@ -107,7 +117,12 @@ describe("remote memory growth", () => {
);
it("does not queue on local exhaustion and honors cancellation before an available growth", async () => {
- const governor = new MemoryGovernor({ limit: 1024, remote: false, monitor: false });
+ const governor = new MemoryGovernor({
+ limit: 1024,
+ remote: false,
+ monitor: false,
+ enabled: true,
+ });
const lease = governor.tryLease(512)!;
await expect(lease.tryGrowAsync(2048)).resolves.toBe(false);
await expect(lease.tryGrowAsync(-1)).rejects.toBeInstanceOf(RangeError);
diff --git a/tests/unit/server-memory-resources.test.ts b/tests/unit/server-memory-resources.test.ts
index c3b94800d..af6150292 100644
--- a/tests/unit/server-memory-resources.test.ts
+++ b/tests/unit/server-memory-resources.test.ts
@@ -106,6 +106,59 @@ describe("内存与 swap 的实际剩余容量", () => {
});
});
+ it("cgroup v2 用量扣除可回收的 inactive file cache，祖先组同样扣除", () => {
+ const snapshot = fixture({
+ "/sys/fs/cgroup/service/worker/memory.max": String(GiB),
+ "/sys/fs/cgroup/service/worker/memory.current": String(1000 * MiB),
+ "/sys/fs/cgroup/service/worker/memory.stat": `anon ${20 * MiB}\nfile ${950 * MiB}\nactive_file ${50 * MiB}\ninactive_file ${900 * MiB}\n`,
+ "/sys/fs/cgroup/service/memory.max": String(2 * GiB),
+ "/sys/fs/cgroup/service/memory.current": String(1900 * MiB),
+ "/sys/fs/cgroup/service/memory.stat": `anon ${900 * MiB}\ninactive_file ${900 * MiB}\n`,
+ });
+ // 只有 inactive 部分视为可回收，active file 仍计入占用。
+ // worker：1024 - (1000 - 900) = 924 MiB；祖先：2048 - (1900 - 900) = 1048 MiB。
+ expect(snapshot.availableRamBytes).toBe(924 * MiB);
+ expect(createMemoryPlan({ env: {}, snapshot }).hotBudgetBytes).toBe(
+ Math.floor(0.6 * (924 - 256) * MiB)
+ );
+ });
+
+ it("cgroup v2 缺少 memory.stat 或字段时按 memory.current 计算", () => {
+ const files = {
+ "/sys/fs/cgroup/service/worker/memory.max": String(GiB),
+ "/sys/fs/cgroup/service/worker/memory.current": String(768 * MiB),
+ };
+ expect(fixture(files).availableRamBytes).toBe(256 * MiB);
+ expect(
+ fixture({ ...files, "/sys/fs/cgroup/service/worker/memory.stat": "anon 1\nfile 2\n" })
+ .availableRamBytes
+ ).toBe(256 * MiB);
+ });
+
+ it("inactive file 大于用量时实际占用不为负", () => {
+ expect(
+ fixture({
+ "/sys/fs/cgroup/service/worker/memory.max": String(GiB),
+ "/sys/fs/cgroup/service/worker/memory.current": String(100 * MiB),
+ "/sys/fs/cgroup/service/worker/memory.stat": `inactive_file ${200 * MiB}\n`,
+ }).availableRamBytes
+ ).toBe(GiB);
+ });
+
+ it("cgroup v1 用量与 memsw 联合用量都扣除 total_inactive_file", () => {
+ expect(
+ fixture({
+ "/proc/self/cgroup": "5:memory:/service",
+ "/proc/self/mountinfo": "29 23 0:26 / /sys/fs/cgroup/memory rw - cgroup cgroup rw,memory",
+ "/sys/fs/cgroup/memory/service/memory.limit_in_bytes": String(2 * GiB),
+ "/sys/fs/cgroup/memory/service/memory.usage_in_bytes": String(1.75 * GiB),
+ "/sys/fs/cgroup/memory/service/memory.stat": `inactive_file ${100 * MiB}\ntotal_inactive_file ${768 * MiB}\n`,
+ "/sys/fs/cgroup/memory/service/memory.memsw.limit_in_bytes": String(2.5 * GiB),
+ "/sys/fs/cgroup/memory/service/memory.memsw.usage_in_bytes": String(2 * GiB),
+ })
+ ).toMatchObject({ availableRamBytes: GiB, availableSwapBytes: 256 * MiB });
+ });
+
it("cgroup namespace 中的根成员仍遵守挂载根限制", () => {
expect(
fixture({
diff --git a/tests/unit/settings/system-settings-form-memory-admission-toggle.test.tsx b/tests/unit/settings/system-settings-form-memory-admission-toggle.test.tsx
new file mode 100644
index 000000000..081dc8297
--- /dev/null
+++ b/tests/unit/settings/system-settings-form-memory-admission-toggle.test.tsx
@@ -0,0 +1,190 @@
+import fs from "node:fs";
+import path from "node:path";
+import type { ReactNode } from "react";
+import { act } from "react";
+import { createRoot } from "react-dom/client";
+import { NextIntlClientProvider } from "next-intl";
+import { beforeEach, describe, expect, test, vi } from "vitest";
+import { SystemSettingsForm } from "@/app/[locale]/settings/config/_components/system-settings-form";
+import type { SystemSettings } from "@/types/system-config";
+
+vi.mock("next/navigation", () => ({
+ useRouter: () => ({ refresh: vi.fn() }),
+}));
+
+const systemConfigActionMocks = vi.hoisted(() => ({
+ saveSystemSettings: vi.fn(async (): Promise => ({ ok: true })),
+}));
+vi.mock("@/actions/system-config", () => systemConfigActionMocks);
+
+const requestFiltersActionMocks = vi.hoisted(() => ({
+ getDistinctProviderGroupsAction: vi.fn(async () => ({ ok: true, data: [] })),
+}));
+vi.mock("@/actions/request-filters", () => requestFiltersActionMocks);
+
+const sonnerMocks = vi.hoisted(() => ({
+ toast: {
+ success: vi.fn(),
+ error: vi.fn(),
+ warning: vi.fn(),
+ },
+}));
+vi.mock("sonner", () => sonnerMocks);
+
+const baseSettings = {
+ siteTitle: "CC Hub",
+ allowGlobalUsageView: true,
+ currencyDisplay: "USD",
+ billingModelSource: "original",
+ codexPriorityBillingSource: "requested",
+ timezone: "UTC",
+ verboseProviderError: false,
+ passThroughUpstreamErrorMessage: true,
+ enableHttp2: true,
+ enableHighConcurrencyMode: false,
+ enableMemoryAdmission: false,
+ interceptAnthropicWarmupRequests: false,
+ enableThinkingSignatureRectifier: true,
+ enableThinkingBudgetRectifier: true,
+ enableBillingHeaderRectifier: true,
+ enableResponseInputRectifier: true,
+ enableCodexSessionIdCompletion: true,
+ enableClaudeMetadataUserIdInjection: true,
+ enableResponseFixer: true,
+ allowNonConversationEndpointProviderFallback: true,
+ fakeStreamingWhitelist: [],
+ responseFixerConfig: {
+ fixEncoding: true,
+ fixSseFormat: true,
+ fixTruncatedJson: true,
+ },
+ quotaDbRefreshIntervalSeconds: 10,
+ quotaLeasePercent5h: 0.05,
+ quotaLeasePercentDaily: 0.05,
+ quotaLeasePercentWeekly: 0.05,
+ quotaLeasePercentMonthly: 0.05,
+ quotaLeaseCapUsd: null,
+ ipGeoLookupEnabled: true,
+ ipExtractionConfig: null,
+ replayEnabled: null,
+ replayCacheTtlMinutes: 30,
+ cacheEffectivenessEnabled: null,
+} satisfies Partial;
+
+function loadMessages(locale: string) {
+ const base = path.join(process.cwd(), `messages/${locale}/settings`);
+ const read = (name: string) => JSON.parse(fs.readFileSync(path.join(base, name), "utf8"));
+
+ return {
+ settings: {
+ common: read("common.json"),
+ config: read("config.json"),
+ requestFilters: read("requestFilters.json"),
+ },
+ };
+}
+
+function render(node: ReactNode) {
+ const container = document.createElement("div");
+ document.body.appendChild(container);
+ const root = createRoot(container);
+
+ act(() => {
+ root.render(
+
+ {node}
+
+ );
+ });
+
+ return {
+ unmount: () => {
+ act(() => root.unmount());
+ container.remove();
+ },
+ };
+}
+
+async function submitForm() {
+ const form = document.body.querySelector("form");
+ if (!form) throw new Error("未找到系统设置表单");
+
+ await act(async () => {
+ form.dispatchEvent(new Event("submit", { bubbles: true, cancelable: true }));
+ await Promise.resolve();
+ await new Promise((resolve) => setTimeout(resolve, 0));
+ });
+}
+
+function memoryAdmissionSwitch(): HTMLElement {
+ const label = loadMessages("en").settings.config.form.enableMemoryAdmission as string;
+ const element = document.body.querySelector(`[aria-label="${label}"]`);
+ if (!element) throw new Error("未找到内存准入开关");
+ return element;
+}
+
+describe("SystemSettingsForm 内存准入开关", () => {
+ beforeEach(() => {
+ document.body.innerHTML = "";
+ vi.clearAllMocks();
+ });
+
+ test("默认关闭，未切换时保存为 false", async () => {
+ const { unmount } = render(
+
+ );
+
+ const toggle = memoryAdmissionSwitch();
+ expect(toggle.id).toBe("enable-memory-admission");
+ expect(toggle.getAttribute("aria-checked")).toBe("false");
+ expect(document.body.textContent).toContain(
+ loadMessages("en").settings.config.form.enableMemoryAdmissionDesc
+ );
+
+ await submitForm();
+
+ expect(systemConfigActionMocks.saveSystemSettings).toHaveBeenCalledWith(
+ expect.objectContaining({ enableMemoryAdmission: false })
+ );
+
+ unmount();
+ });
+
+ test("开启后保存为 true，并按服务端返回值刷新开关", async () => {
+ systemConfigActionMocks.saveSystemSettings.mockResolvedValueOnce({
+ ok: true,
+ data: { ...baseSettings, enableMemoryAdmission: false },
+ });
+ const { unmount } = render(
+
+ );
+
+ await act(async () => {
+ memoryAdmissionSwitch().click();
+ await Promise.resolve();
+ });
+ expect(memoryAdmissionSwitch().getAttribute("aria-checked")).toBe("true");
+
+ await submitForm();
+
+ expect(systemConfigActionMocks.saveSystemSettings).toHaveBeenCalledWith(
+ expect.objectContaining({ enableMemoryAdmission: true })
+ );
+ expect(memoryAdmissionSwitch().getAttribute("aria-checked")).toBe("false");
+
+ unmount();
+ });
+
+ test("初始值为开启时显示为开启", () => {
+ const { unmount } = render(
+
+ );
+
+ expect(memoryAdmissionSwitch().getAttribute("aria-checked")).toBe("true");
+
+ unmount();
+ });
+});