diff --git a/.env.example b/.env.example index 858324ad9f..91fe676c4c 100644 --- a/.env.example +++ b/.env.example @@ -1001,6 +1001,11 @@ APP_LOG_TO_FILE=true # Default: 100000 # CALL_LOGS_TABLE_MAX_ROWS=100000 +# Maximum age for orphaned active request log entries before the in-memory +# pending-request reaper removes them. Accepts milliseconds. +# Default: 3600000 (1 hour) +# MAX_PENDING_REQUEST_AGE_MS=3600000 + # Whether call log pipeline capture stores stream chunks when enabled in settings. # Only applies when call_log_pipeline_enabled=true. # Default: true diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index ad5590de62..c8d791b781 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -634,6 +634,7 @@ The logging system writes to both stdout and rotated log files. All configuratio | `CALL_LOG_RETENTION_DAYS` | `7` | Days to keep request/call log entries in the database. | | `CALL_LOG_MAX_ENTRIES` | `10000` | Max call log entries in the in-memory buffer. | | `CALL_LOGS_TABLE_MAX_ROWS` | `100000` | Max rows in the `call_logs` SQLite table before pruning. | +| `MAX_PENDING_REQUEST_AGE_MS` | `3600000` (1 hour) | Max age for orphaned active request log entries before in-memory cleanup. | | `CALL_LOG_PIPELINE_CAPTURE_STREAM_CHUNKS` | `true` | Store stream chunks in pipeline artifacts when `call_log_pipeline_enabled=true`. | | `CALL_LOG_PIPELINE_MAX_SIZE_KB` | `512` | Max pipeline call log artifact size in KB when `call_log_pipeline_enabled=true`. | | `PROXY_LOGS_TABLE_MAX_ROWS` | `100000` | Max rows in the `proxy_logs` SQLite table before pruning. | diff --git a/src/lib/usage/usageHistory.ts b/src/lib/usage/usageHistory.ts index ae7772be7a..599619eb0a 100644 --- a/src/lib/usage/usageHistory.ts +++ b/src/lib/usage/usageHistory.ts @@ -215,26 +215,18 @@ const pendingRequests: { */ const pendingById = new Map(); -/** - * Orphaned-pending-request reaper. - * - * Pending details are normally removed when a request finalizes (clean completion, - * tracked error, client cancel). But a request that never finalizes cleanly — an - * upstream/fetch error thrown before the finalize call, a client disconnect, or a - * process-level timeout — leaves its detail in `pendingById` (and `pendingRequests.details`) - * forever, each retaining truncated request/response payload previews. Under real proxy - * traffic a steady fraction of requests terminate abnormally, so this grows monotonically - * (previously only an admin reset via clearPendingRequests() could free it). - * - * The reaper drops entries whose `startedAt` is older than MAX_PENDING_REQUEST_AGE_MS — far - * longer than any genuine request — so only truly-orphaned entries are evicted; a live - * request always finalizes long before that. MAX_PENDING_DETAILS is a hard backstop. - */ -const MAX_PENDING_REQUEST_AGE_MS = 15 * 60 * 1000; +const DEFAULT_MAX_PENDING_REQUEST_AGE_MS = 60 * 60 * 1000; const MAX_PENDING_DETAILS = 5000; const PENDING_SWEEP_INTERVAL_MS = 5 * 60 * 1000; let _pendingSweepTimer: ReturnType | null = null; +export function getMaxPendingRequestAgeMs( + rawValue: string | undefined = process.env.MAX_PENDING_REQUEST_AGE_MS +): number { + const parsed = Number.parseInt(rawValue ?? "", 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_MAX_PENDING_REQUEST_AGE_MS; +} + function ensurePendingSweepTimer(): void { if (_pendingSweepTimer || typeof setInterval !== "function") return; _pendingSweepTimer = setInterval(() => { @@ -256,7 +248,7 @@ function ensurePendingSweepTimer(): void { */ export function sweepStalePendingRequests( now: number = Date.now(), - maxAgeMs: number = MAX_PENDING_REQUEST_AGE_MS + maxAgeMs: number = getMaxPendingRequestAgeMs() ): number { let removed = 0; @@ -347,12 +339,13 @@ export function trackPendingRequest( if (!pendingRequests.details[connectionId][modelKey]) { pendingRequests.details[connectionId][modelKey] = []; } + const now = Date.now(); const newDetail = { - id: `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, + id: `${now}-${Math.random().toString(36).slice(2, 8)}`, model, provider, connectionId, - startedAt: Date.now(), + startedAt: now, ...normalizedMetadata, }; pendingRequests.details[connectionId][modelKey].push(newDetail); diff --git a/tests/unit/usage-pending-sweep.test.ts b/tests/unit/usage-pending-sweep.test.ts index af7ed8b085..d3de475b16 100644 --- a/tests/unit/usage-pending-sweep.test.ts +++ b/tests/unit/usage-pending-sweep.test.ts @@ -1,52 +1,114 @@ -import test from "node:test"; -import assert from "node:assert/strict"; - -const { - trackPendingRequest, - getPendingById, - getPendingRequests, - sweepStalePendingRequests, - clearPendingRequests, -} = await import("../../src/lib/usage/usageHistory.ts"); - -test("sweepStalePendingRequests evicts orphaned pending details and self-heals counts", () => { - clearPendingRequests(); - - // One request that will be treated as orphaned (never finalized), one fresh. - const staleId = trackPendingRequest("gpt-x", "openai", "conn-stale", true); - const freshId = trackPendingRequest("gpt-x", "openai", "conn-fresh", true); - - assert.ok(staleId && freshId, "both started requests should produce ids"); - assert.equal(getPendingById().size, 2); - assert.equal(getPendingRequests().byModel["gpt-x (openai)"], 2); - - // Age the stale entry well beyond the max age. - const stale = getPendingById().get(staleId); - assert.ok(stale, "stale detail should exist"); - stale.startedAt = Date.now() - 60 * 60 * 1000; // 1 hour ago - - const removed = sweepStalePendingRequests(Date.now(), 15 * 60 * 1000); - - assert.equal(removed, 1, "exactly one orphaned entry should be swept"); - assert.equal(getPendingById().size, 1, "only the fresh entry should remain"); - assert.ok(getPendingById().has(freshId), "fresh entry must survive"); - - // Counts must reflect the eviction (decremented, not left dangling). - assert.equal(getPendingRequests().byModel["gpt-x (openai)"], 1); - assert.equal(getPendingRequests().byAccount["conn-stale"], undefined); - assert.equal(getPendingRequests().byAccount["conn-fresh"]["gpt-x (openai)"], 1); - - clearPendingRequests(); -}); - -test("sweepStalePendingRequests is a no-op when nothing is stale", () => { - clearPendingRequests(); - trackPendingRequest("m", "p", "c1", true); - trackPendingRequest("m", "p", "c2", true); - - const removed = sweepStalePendingRequests(Date.now(), 15 * 60 * 1000); - - assert.equal(removed, 0); - assert.equal(getPendingById().size, 2); - clearPendingRequests(); -}); +import test from "node:test"; +import assert from "node:assert/strict"; + +const { + trackPendingRequest, + getPendingById, + getPendingRequests, + sweepStalePendingRequests, + getMaxPendingRequestAgeMs, + clearPendingRequests, +} = await import("../../src/lib/usage/usageHistory.ts"); + +const MINUTE_MS = 60 * 1000; +const HOUR_MS = 60 * MINUTE_MS; + +test("sweepStalePendingRequests evicts orphaned pending details and self-heals counts", () => { + clearPendingRequests(); + + // One request that will be treated as orphaned (never finalized), one fresh. + const staleId = trackPendingRequest("gpt-x", "openai", "conn-stale", true); + const freshId = trackPendingRequest("gpt-x", "openai", "conn-fresh", true); + + assert.ok(staleId && freshId, "both started requests should produce ids"); + assert.equal(getPendingById().size, 2); + assert.equal(getPendingRequests().byModel["gpt-x (openai)"], 2); + + // Age the stale entry well beyond the max age. + const stale = getPendingById().get(staleId); + assert.ok(stale, "stale detail should exist"); + stale.startedAt = Date.now() - 2 * HOUR_MS; + + const removed = sweepStalePendingRequests(Date.now(), HOUR_MS); + + assert.equal(removed, 1, "exactly one orphaned entry should be swept"); + assert.equal(getPendingById().size, 1, "only the fresh entry should remain"); + assert.ok(getPendingById().has(freshId), "fresh entry must survive"); + + // Counts must reflect the eviction (decremented, not left dangling). + assert.equal(getPendingRequests().byModel["gpt-x (openai)"], 1); + assert.equal(getPendingRequests().byAccount["conn-stale"], undefined); + assert.equal(getPendingRequests().byAccount["conn-fresh"]["gpt-x (openai)"], 1); + + clearPendingRequests(); +}); + +test("sweepStalePendingRequests is a no-op when nothing is stale", () => { + clearPendingRequests(); + trackPendingRequest("m", "p", "c1", true); + trackPendingRequest("m", "p", "c2", true); + + const removed = sweepStalePendingRequests(Date.now(), HOUR_MS); + + assert.equal(removed, 0); + assert.equal(getPendingById().size, 2); + clearPendingRequests(); +}); + +test("sweepStalePendingRequests defaults to a one hour max pending age", () => { + clearPendingRequests(); + + const staleId = trackPendingRequest("m", "p", "old", true); + const recentId = trackPendingRequest("m", "p", "recent", true); + assert.ok(staleId && recentId); + + const now = Date.now(); + const stale = getPendingById().get(staleId); + const recent = getPendingById().get(recentId); + assert.ok(stale && recent); + + stale.startedAt = now - 61 * MINUTE_MS; + recent.startedAt = now - 59 * MINUTE_MS; + + const removed = sweepStalePendingRequests(now); + + assert.equal(removed, 1); + assert.equal(getPendingById().has(staleId), false); + assert.equal(getPendingById().has(recentId), true); + clearPendingRequests(); +}); + +test("pending sweep max age can be overridden through environment", () => { + clearPendingRequests(); + const previous = process.env.MAX_PENDING_REQUEST_AGE_MS; + process.env.MAX_PENDING_REQUEST_AGE_MS = String(2 * HOUR_MS); + + try { + const requestId = trackPendingRequest("m", "p", "custom-age", true); + assert.ok(requestId); + + const detail = getPendingById().get(requestId); + assert.ok(detail); + detail.startedAt = Date.now() - 90 * MINUTE_MS; + + assert.equal(getMaxPendingRequestAgeMs(), 2 * HOUR_MS); + assert.equal(sweepStalePendingRequests(Date.now()), 0); + assert.equal(getPendingById().has(requestId), true); + } finally { + if (previous === undefined) delete process.env.MAX_PENDING_REQUEST_AGE_MS; + else process.env.MAX_PENDING_REQUEST_AGE_MS = previous; + clearPendingRequests(); + } +}); + +test("invalid pending sweep max age falls back to one hour", () => { + const previous = process.env.MAX_PENDING_REQUEST_AGE_MS; + process.env.MAX_PENDING_REQUEST_AGE_MS = "not-a-number"; + + try { + assert.equal(getMaxPendingRequestAgeMs(), HOUR_MS); + } finally { + if (previous === undefined) delete process.env.MAX_PENDING_REQUEST_AGE_MS; + else process.env.MAX_PENDING_REQUEST_AGE_MS = previous; + } +});