Skip to content

fix(proxy): prevent large session requests from cascading into OOM #1457

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ding113 merged 3 commits into from
Aug 27, 2026
Merged

fix(proxy): prevent large session requests from cascading into OOM #1457

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
80feea2
fix(session): bound request debug artifact storage to prevent OOM
ding113 Aug 27, 2026
8beb34d
perf(active-sessions): prefer snapshots over legacy payload reads
ding113 Aug 27, 2026
ad00c39
fix(forwarder): cap legacy streaming hedge concurrency at two
ding113 Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .env.example
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,8 @@ SESSION_RESPONSE_BODY_DEDUP_ENABLED=false # response body 单 key 去重 writer
SESSION_RESPONSE_BODY_MAX_BYTES=5242880 # 会话响应体 Redis 存储上限（默认 5 MiB，范围 64 KiB-64 MiB）
# 去重关闭时限制每份正文；去重开启时限制所有唯一正文的 UTF-8 总字节数
# 超限正文不落 Redis；before/after snapshot 的 headers/meta 仍保留
SESSION_REQUEST_ARTIFACT_MAX_BYTES=5242880 # 单份请求调试正文 Redis 存储上限（默认 5 MiB，范围 64 KiB-64 MiB）
# 超限时跳过 requestBody/messages 和 before/after body，保留 headers/meta

# Dashboard 配置
DASHBOARD_LOGS_POLL_INTERVAL_MS=5000 # 日志页自动刷新轮询间隔（毫秒，默认 5000，范围 250-60000）
Expand Down
64 changes: 36 additions & 28 deletions src/actions/active-sessions.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1029,11 +1029,8 @@ export async function getSessionDetails(
locatorResult.locator.keyId
);

// 6. 并行获取 messages、requestBody 和 response（不缓存，因为这些数据较大）
// 6. 先读取 phase 快照和轻量 metadata；大字段仅在现代快照缺失时读取 legacy key。
const [
requestBody,
messages,
response,
requestHeaders,
responseHeaders,
clientReqMeta,
Expand All @@ -1046,15 +1043,6 @@ export async function getSessionDetails(
responseSnapshotBefore,
responseSnapshotAfter,
] = await Promise.all([
redisArtifactsOwned
? SessionManager.getSessionRequestBody(sourceSessionId, effectiveSequence)
: null,
redisArtifactsOwned
? SessionManager.getSessionMessages(sourceSessionId, effectiveSequence)
: null,
redisArtifactsOwned
? SessionManager.getSessionResponse(sourceSessionId, effectiveSequence)
: null,
redisArtifactsOwned
? SessionManager.getSessionRequestHeaders(sourceSessionId, effectiveSequence)
: null,
Expand Down Expand Up @@ -1100,21 +1088,6 @@ export async function getSessionDetails(
: null,
]);

// 兼容：历史/异常数据可能是 JSON 字符串（前端需要根级对象/数组）
const normalizedMessages = parseJsonStringOrNull(messages);
const normalizedRequestBody = parseJsonStringOrNull(requestBody);

const requestMeta = {
clientUrl: clientReqMeta?.url ?? null,
upstreamUrl: upstreamReqMeta?.url ?? null,
method: clientReqMeta?.method ?? upstreamReqMeta?.method ?? null,
};

const responseMeta = {
upstreamUrl: upstreamResMeta?.url ?? upstreamReqMeta?.url ?? null,
statusCode: upstreamResMeta?.statusCode ?? null,
};

const snapshots: SessionDetailSnapshots = {
defaultView: DEFAULT_SESSION_DETAIL_VIEW_MODE,
request: {
Expand All @@ -1134,6 +1107,41 @@ export async function getSessionDetails(
after: normalizeResponseSnapshot(responseSnapshotAfter ?? null),
},
};

const snapshotRequestBody = snapshots.request.after?.body ?? snapshots.request.before?.body;

@github-actions github-actions Bot Aug 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[MEDIUM] [LOGIC-BUG] Legacy requestBody field now prefers the upstream-forwarded body over the client body

Why this is a problem: The legacy key this field mirrors (storeSessionRequestBody / getSessionRequestBody) is documented as "客户端原始请求体" - the client's request body captured at guard time, before any format conversion or request filtering. This derivation prefers snapshots.request.after.body, which is session.forwardedRequestBody: the post-conversion body actually sent to the upstream provider. For cross-format routing (e.g. an OpenAI client served by a Claude provider) or model-redirected requests, data.requestBody returned by GET /api/v1/sessions/{id} now contains a completely different payload than it did before this PR, while both snapshots still exist (the normal case for completed requests). The adjacent snapshotMessages derivation two lines below prefers before first, correctly preserving the client-side semantics of its legacy counterpart - both legacy fields were captured at the same guard-time point, so requestBody should match that ordering.

Suggested fix:

const snapshotRequestBody = snapshots.request.before?.body ?? snapshots.request.after?.body;

The forwarded body remains available to consumers via data.snapshots.request.after, which is what the dashboard UI reads (session-messages-client.tsx only consumes snapshots).

const snapshotMessages =
snapshots.request.before?.messages ?? snapshots.request.after?.messages;
const snapshotResponse = snapshots.response.after?.body ?? snapshots.response.before?.body;

const [legacyRequestBody, legacyMessages, legacyResponse] = await Promise.all([
redisArtifactsOwned && snapshotRequestBody == null
? SessionManager.getSessionRequestBody(sourceSessionId, effectiveSequence)
: null,
redisArtifactsOwned && snapshotMessages == null
? SessionManager.getSessionMessages(sourceSessionId, effectiveSequence)
: null,
redisArtifactsOwned && snapshotResponse == null
? SessionManager.getSessionResponse(sourceSessionId, effectiveSequence)
: null,
]);

// 兼容：历史/异常数据可能是 JSON 字符串（前端需要根级对象/数组）。
const normalizedMessages = snapshotMessages ?? parseJsonStringOrNull(legacyMessages ?? null);
const normalizedRequestBody =
snapshotRequestBody ?? parseJsonStringOrNull(legacyRequestBody ?? null);
const response = snapshotResponse ?? legacyResponse;

const requestMeta = {
clientUrl: clientReqMeta?.url ?? null,
upstreamUrl: upstreamReqMeta?.url ?? null,
method: clientReqMeta?.method ?? upstreamReqMeta?.method ?? null,
};

const responseMeta = {
upstreamUrl: upstreamResMeta?.url ?? upstreamReqMeta?.url ?? null,
statusCode: upstreamResMeta?.statusCode ?? null,
};

const legacyCompatibilitySnapshots = buildLegacyCompatibilitySnapshots({
requestBody: normalizedRequestBody,
messages: normalizedMessages,
Expand Down
2 changes: 2 additions & 0 deletions src/app/v1/_lib/proxy/forwarder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,7 @@ import {
export const DEFAULT_CODEX_USER_AGENT =
"codex_cli_rs/0.93.0 (Windows 10.0.26200; x86_64) vscode/1.108.1";
const EMPTY_PREFIX_CHUNK = new Uint8Array(0);
const LEGACY_STREAMING_HEDGE_MAX_CONCURRENCY = 2;

async function runStreamContentGateWithAbortSignals(
reader: ReadableStreamDefaultReader<Uint8Array>,
Expand Down Expand Up @@ -4957,6 +4958,7 @@ export class ProxyForwarder {

const launchAlternative = async () => {
if (settled || winnerCommitted || noMoreProviders) return;
if (attempts.size >= LEGACY_STREAMING_HEDGE_MAX_CONCURRENCY) return;
if (launchingAlternative) {
await launchingAlternative;
return;
Expand Down
37 changes: 25 additions & 12 deletions src/app/v1/_lib/proxy/session-guard.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,15 @@ export class ProxySessionGuard {
session.setRawCrossProviderFallbackEnabled(rawFallbackEnabled);
const allowRawSessionContext = session.isRawCrossProviderFallbackEnabled();
session.setHighConcurrencyModeEnabled(systemSettings.enableHighConcurrencyMode ?? false);
const persistSessionRequestArtifacts =
session.shouldPersistSessionDebugArtifacts() &&
session.shouldPersistSessionRequestArtifacts();
let requestMessageBeforeProxyMutations = session.request.message as Record<string, unknown>;
if (session.request.message && typeof session.request.message === "object") {
if (
persistSessionRequestArtifacts &&
session.request.message &&
typeof session.request.message === "object"
) {
try {
requestMessageBeforeProxyMutations = structuredClone(
session.request.message as Record<string, unknown>
Expand All @@ -99,7 +106,7 @@ export class ProxySessionGuard {
requestMessageBeforeProxyMutations = session.request.message as Record<string, unknown>;
}
}
const originalMessages = session.getMessages();
const originalMessages = persistSessionRequestArtifacts ? session.getMessages() : undefined;

// Codex Session ID 补全：在提取 clientSessionId 之前触发，避免落入不稳定的降级方案
const codexCompletionEnabled = systemSettings.enableCodexSessionIdCompletion ?? true;
Expand Down Expand Up @@ -232,23 +239,29 @@ export class ProxySessionGuard {
// 注意：必须在后续任何格式转换/过滤前触发存储，避免记录被“后处理”污染
if (session.sessionId && session.shouldPersistSessionDebugArtifacts()) {
const requestBeforeSnapshot = {
body: requestMessageBeforeProxyMutations,
headers: filterClientRequestSnapshotHeaders(session.headers),
meta: {
clientUrl: session.requestUrl.toString(),
upstreamUrl: null,
method: session.method,
},
...(originalMessages !== undefined ? { messages: originalMessages } : {}),
...(persistSessionRequestArtifacts
? {
body: requestMessageBeforeProxyMutations,
...(originalMessages !== undefined ? { messages: originalMessages } : {}),
}
: {}),
};

void SessionManager.storeSessionRequestBody(
session.sessionId,
session.request.message,
requestSequence
).catch((err) => {
logger.error("[ProxySessionGuard] Failed to store session request body:", err);
});
if (persistSessionRequestArtifacts) {
void SessionManager.storeSessionRequestBody(
session.sessionId,
session.request.message,
requestSequence
).catch((err) => {
logger.error("[ProxySessionGuard] Failed to store session request body:", err);
});
}

void SessionManager.storeSessionClientRequestMeta(
session.sessionId,
Expand All @@ -270,7 +283,7 @@ export class ProxySessionGuard {
});

// 可选：存储 messages（受环境变量控制，按请求序号独立存储）
if (messages !== undefined) {
if (persistSessionRequestArtifacts && messages !== undefined) {
void SessionManager.storeSessionMessages(
session.sessionId,
messages,
Expand Down
20 changes: 20 additions & 0 deletions src/app/v1/_lib/proxy/session.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@ import {
writeLiveRoutingTrace,
} from "@/lib/redis/live-chain-store";
import type { SessionBindingSnapshot } from "@/lib/redis/session-binding";
import {
getSessionRequestArtifactByteSize,
getSessionRequestArtifactMaxBytes,
} from "@/lib/session-request-artifact-limit";
import { clientRequestsContext1m as clientRequestsContext1mHelper } from "@/lib/special-attributes";
import { ERROR_CODES, getErrorMessageServer } from "@/lib/utils/error-messages";
import {
Expand Down Expand Up @@ -581,6 +585,22 @@ export class ProxySession {
return !this.highConcurrencyModeEnabled;
}

shouldPersistSessionRequestArtifacts(): boolean {

@github-actions github-actions Bot Aug 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[MEDIUM] [TEST-MISSING-CRITICAL] No unit test covers the artifact-size gate's measurement logic

Why this is a problem: This method contains the boundary logic of the OOM fix itself - buffer-hint precedence (avoiding re-serialization of the body), the multipart exception (measuring the small logical body instead of the raw multipart buffer, which would falsely reject image requests), and the byteSize <= maxBytes cutoff. Every test in this PR bypasses the real implementation: session-guard-warmup-intercept.test.ts stubs shouldPersistSessionRequestArtifacts to a constant (lines 97 and 307), and the session-manager tests exercise only the string path of getSessionRequestArtifactByteSize. A regression here - for example dropping the multipart exception so every image request is measured via JSON.stringify, or inverting the hint precedence so the buffer is ignored - would silently re-introduce the serialization spike this PR eliminates, and the suite would still pass. CLAUDE.md requires "All new features must have unit test coverage of at least 80%", and the sibling gate shouldPersistSessionDebugArtifacts is already directly tested in tests/unit/proxy/session.test.ts ("ProxySession high-concurrency policy").

Suggested fix (add to tests/unit/proxy/session.test.ts, reusing its createSession helper):

describe("ProxySession request artifact gate", () => {
  it("buffer 提示超过上限时跳过 request artifacts", () => {
    const session = createSession({ redirectedModel: null, requestMessage: { messages: [] } });
    session.request.buffer = new Uint8Array(6 * 1024 * 1024);
    expect(session.shouldPersistSessionRequestArtifacts()).toBe(false);
  });

  it("multipart 请求按逻辑 body 而非原始 buffer 判定大小", () => {
    const session = createSession({
      redirectedModel: null,
      requestMessage: { model: "gpt-image-1", prompt: "cat" },
    });
    session.request.buffer = new Uint8Array(6 * 1024 * 1024);
    vi.spyOn(session, "isOpenAIImageMultipartRequest").mockReturnValue(true);
    expect(session.shouldPersistSessionRequestArtifacts()).toBe(true);
  });

  it("恰好等于上限时仍保留 artifacts", () => {
    const session = createSession({ redirectedModel: null, requestMessage: { messages: [] } });
    session.request.buffer = new Uint8Array(5 * 1024 * 1024);
    expect(session.shouldPersistSessionRequestArtifacts()).toBe(true);
  });
});

const byteSize = getSessionRequestArtifactByteSize(
this.request.message,
this.isOpenAIImageMultipartRequest() ? undefined : this.request.buffer?.byteLength
);
const maxBytes = getSessionRequestArtifactMaxBytes();
if (byteSize <= maxBytes) return true;

logger.warn("[ProxySession] Skipped oversized session request artifacts", {
byteSize,
maxBytes,
endpoint: this.getEndpoint(),
});
return false;
}

shouldTrackSessionObservability(): boolean {
return !this.highConcurrencyModeEnabled;
}
Expand Down
7 changes: 7 additions & 0 deletions src/lib/config/env.schema.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,13 @@ export const EnvSchema = z.object({
// - false (默认)：存储请求/响应体但对 message 内容脱敏 [REDACTED]
// - true：原样存储 message 内容（注意隐私和存储空间影响）
STORE_SESSION_MESSAGES: z.string().default("false").transform(booleanTransform),
// 单份请求调试 artifact（requestBody/messages/before/after body）的 Redis 存储上限。
SESSION_REQUEST_ARTIFACT_MAX_BYTES: z.coerce
.number()
.int()
.min(64 * 1024)
.max(64 * 1024 * 1024)
.default(5 * 1024 * 1024),
// 会话响应体存储开关
// - true (默认)：存储响应体（SSE/JSON），用于调试/回放/问题定位（Redis 临时缓存，默认 5 分钟）
// - false：不存储响应体（注意：不影响本次请求处理；仅影响后续在 UI/诊断中查看 response body）
Expand Down
41 changes: 41 additions & 0 deletions src/lib/session-manager-detail-snapshots.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,12 +64,14 @@ vi.mock("@/lib/redis", () => ({

let mockStoreMessages = false;
let mockStoreSessionResponseBody = true;
let mockSessionRequestArtifactMaxBytes = 1024 * 1024;
let mockSessionResponseBodyMaxBytes = 1024 * 1024;

vi.mock("@/lib/config/env.schema", () => ({
getEnvConfig: () => ({
STORE_SESSION_MESSAGES: mockStoreMessages,
STORE_SESSION_RESPONSE_BODY: mockStoreSessionResponseBody,
SESSION_REQUEST_ARTIFACT_MAX_BYTES: mockSessionRequestArtifactMaxBytes,
SESSION_RESPONSE_BODY_MAX_BYTES: mockSessionResponseBodyMaxBytes,
SESSION_TTL: 300,
}),
Expand All @@ -84,6 +86,7 @@ describe("SessionManager detail snapshots", () => {
redisMock.status = "ready";
mockStoreMessages = false;
mockStoreSessionResponseBody = true;
mockSessionRequestArtifactMaxBytes = 1024 * 1024;
mockSessionResponseBodyMaxBytes = 1024 * 1024;
});

Expand Down Expand Up @@ -429,6 +432,44 @@ describe("SessionManager detail snapshots", () => {
);
});

it("skips oversized request artifacts while preserving snapshot headers and meta", async () => {
mockStoreMessages = true;
mockSessionRequestArtifactMaxBytes = 4;

await SessionManager.storeSessionRequestBody("sess_oversized_request", "12345", 1);
await SessionManager.storeSessionMessages("sess_oversized_request", ["12345"], 1);
await SessionManager.storeSessionRequestPhaseSnapshot(
"sess_oversized_request",
"before",
{
body: "12345",
messages: ["12345"],
headers: new Headers({ "content-type": "application/json" }),
meta: {
clientUrl: "https://client.example/v1/messages",
upstreamUrl: null,
method: "POST",
},
},
1
);

expect(await SessionManager.getSessionRequestBody("sess_oversized_request", 1)).toBeNull();
expect(await SessionManager.getSessionMessages("sess_oversized_request", 1)).toBeNull();
expect(
await SessionManager.getSessionRequestPhaseSnapshot("sess_oversized_request", "before", 1)
).toEqual({
body: null,
messages: null,
headers: { "content-type": "application/json" },
meta: {
clientUrl: "https://client.example/v1/messages",
upstreamUrl: null,
method: "POST",
},
});
});

it("removes a previous snapshot body when its replacement exceeds the limit", async () => {
mockSessionResponseBodyMaxBytes = 4;

Expand Down
Loading
Loading