mirror of
https://github.com/musistudio/claude-code-router.git
synced 2026-08-31 01:35:16 +08:00
Refactor router pipeline and update related components
This commit is contained in:
musistudio
@@ -115,6 +115,7 @@ class GatewayService {
|
||||
return this.status;
|
||||
}
|
||||
|
||||
await this.rawTraceSynchronizer.start();
|
||||
await this.listen(config);
|
||||
if (this.server) {
|
||||
const proxyStatus = await proxyService.attach(config, this.server);
|
||||
@@ -184,6 +185,7 @@ class GatewayService {
|
||||
if (server) {
|
||||
await closeServer(server);
|
||||
}
|
||||
await this.rawTraceSynchronizer.stop();
|
||||
|
||||
await proxyService.stop(options.proxyRestoreTimeoutMs);
|
||||
await pluginService.stop();
|
||||
|
||||
@@ -2,7 +2,11 @@ import { randomUUID } from "node:crypto";
|
||||
import type { IncomingMessage, ServerResponse } from "node:http";
|
||||
import { Readable } from "node:stream";
|
||||
import type { ApiKeyConfig, AppConfig, RequestRouteTraceChange } from "@ccr/core/contracts/app";
|
||||
import { createSseErrorDetector, recordGatewayRequestLog } from "@ccr/core/observability/request-log-store";
|
||||
import {
|
||||
createSseErrorDetector,
|
||||
markGatewayRequestLogDropped,
|
||||
recordGatewayRequestLog
|
||||
} from "@ccr/core/observability/request-log-store";
|
||||
import { recordGatewayUsageCapture, type UsageCaptureInput } from "@ccr/core/usage/store";
|
||||
import { ClaudeCodeRouterPlugin } from "@ccr/core/gateway/claude-code-router-plugin";
|
||||
import { adaptRouteRequestBody, restoreRouteRequestBody } from "@ccr/core/routing/protocol-adapter";
|
||||
@@ -220,11 +224,14 @@ export class GatewayRequestPipeline {
|
||||
const successful = statusCode >= 200 && statusCode < 400 && !error;
|
||||
const successSampleRate = config.observability.requestLogSuccessSampleRate ?? 1;
|
||||
if (successful && !requestLogSampled(requestId, successSampleRate)) {
|
||||
markGatewayRequestLogDropped(requestId, "sampled");
|
||||
return;
|
||||
}
|
||||
const bodyCapture = config.observability.requestLogBodyCapture ?? "all";
|
||||
const captureBody = bodyCapture === "all" || (bodyCapture === "errors" && !successful);
|
||||
recordGatewayRequestLog({
|
||||
captureBody: bodyCapture === "all" || (bodyCapture === "errors" && !successful),
|
||||
bodyCapturePolicy: bodyCapture,
|
||||
captureBody,
|
||||
client,
|
||||
completedAt: new Date().toISOString(),
|
||||
durationMs: Date.now() - startedAt,
|
||||
@@ -239,7 +246,7 @@ export class GatewayRequestPipeline {
|
||||
requestBody: shouldSendBody(method) ? bodyToForward ?? Buffer.alloc(0) : Buffer.alloc(0),
|
||||
requestHeaders: headers,
|
||||
requestId,
|
||||
routeTrace: routeTrace?.finish(),
|
||||
routeTrace: routeTrace?.finish({ captureBodyValues: captureBody }),
|
||||
responseBodyText,
|
||||
responseBodySizeBytes,
|
||||
responseBodyTruncated,
|
||||
|
||||
@@ -300,10 +300,16 @@ export async function fetchUpstreamWithFallback(input: {
|
||||
});
|
||||
const hasNextAttempt = index < attempts.length - 1;
|
||||
const attemptUrl = rewriteRouteModelInUrl(input.upstreamUrl, attempt.model);
|
||||
const attemptHeaders = withCoreGatewayAuthHeader(
|
||||
omitLocalObservabilityHeaders(attempt.headers ?? input.headers),
|
||||
input.coreAuthToken
|
||||
);
|
||||
const attemptHeaders = {
|
||||
...withCoreGatewayAuthHeader(
|
||||
omitLocalObservabilityHeaders(attempt.headers ?? input.headers),
|
||||
input.coreAuthToken
|
||||
),
|
||||
// Core raw traces use a unique request id for every fallback attempt,
|
||||
// while turnKey identifies the outer gateway request. Keep both and mark
|
||||
// the attempt so only the final response may refine the stored outcome.
|
||||
"x-ccr-route-attempt": String(attemptNumber)
|
||||
};
|
||||
const attemptProvider = attempt.logicalProvider ?? (
|
||||
attempt.target?.kind === "provider" ? attempt.target.provider.name : undefined
|
||||
);
|
||||
|
||||
@@ -160,6 +160,11 @@ export function bundledToolHubMcpEntryPathCandidates(): string[] {
|
||||
pathJoin(resourcesPath, "app", "dist", "main", TOOL_HUB_MCP_RUNTIME_FILE_NAME)
|
||||
]
|
||||
: []),
|
||||
// Core tests bundle runtime entry points here without producing any
|
||||
// packages/*/dist artifacts. Do not prefer test artifacts in normal runs.
|
||||
...(process.env.NODE_TEST_CONTEXT
|
||||
? [pathJoin(process.cwd(), ".test-dist", "core", "runtime", TOOL_HUB_MCP_RUNTIME_FILE_NAME)]
|
||||
: []),
|
||||
pathJoin(process.cwd(), "packages", "electron", "dist", "main", TOOL_HUB_MCP_RUNTIME_FILE_NAME),
|
||||
pathJoin(process.cwd(), "packages", "cli", "dist", "main", TOOL_HUB_MCP_RUNTIME_FILE_NAME),
|
||||
pathJoin(process.cwd(), "packages", "core", "dist", "main", TOOL_HUB_MCP_RUNTIME_FILE_NAME),
|
||||
|
||||
@@ -44,13 +44,50 @@ let catalog: PriceCatalog | undefined;
|
||||
let catalogPromise: Promise<PriceCatalog> | undefined;
|
||||
|
||||
export async function estimateUsageCostUsd(input: UsageCostInput): Promise<UsageCostEstimate | undefined> {
|
||||
if (!hasBillableUsage(input)) {
|
||||
return undefined;
|
||||
}
|
||||
const model = input.model?.trim();
|
||||
if (!model || model === "unknown") {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const prices = await getPriceCatalog();
|
||||
const price = findModelPrice(prices.index, model, input.provider);
|
||||
return estimateUsageCostFromIndex(input, prices.index, model);
|
||||
}
|
||||
|
||||
/**
|
||||
* Loads the remote catalog without holding a caller's database transaction.
|
||||
* Callers that own a write transaction can then use the cache-only estimator.
|
||||
*/
|
||||
export async function preloadUsagePriceCatalog(): Promise<void> {
|
||||
await getPriceCatalog();
|
||||
}
|
||||
|
||||
export function usagePriceCatalogNeedsRefresh(): boolean {
|
||||
return !catalog || Date.now() - catalog.loadedAt >= catalogTtlMs;
|
||||
}
|
||||
|
||||
/** Never performs I/O. Returns undefined when no fresh catalog is loaded. */
|
||||
export function estimateUsageCostUsdFromLoadedCatalog(
|
||||
input: UsageCostInput
|
||||
): UsageCostEstimate | undefined {
|
||||
if (!hasBillableUsage(input)) {
|
||||
return undefined;
|
||||
}
|
||||
const model = input.model?.trim();
|
||||
if (!model || model === "unknown" || usagePriceCatalogNeedsRefresh() || !catalog) {
|
||||
return undefined;
|
||||
}
|
||||
return estimateUsageCostFromIndex(input, catalog.index, model);
|
||||
}
|
||||
|
||||
function estimateUsageCostFromIndex(
|
||||
input: UsageCostInput,
|
||||
index: PriceIndex,
|
||||
model: string
|
||||
): UsageCostEstimate | undefined {
|
||||
const price = findModelPrice(index, model, input.provider);
|
||||
if (!price) {
|
||||
return undefined;
|
||||
}
|
||||
@@ -76,6 +113,13 @@ export async function estimateUsageCostUsd(input: UsageCostInput): Promise<Usage
|
||||
};
|
||||
}
|
||||
|
||||
function hasBillableUsage(input: UsageCostInput): boolean {
|
||||
return normalizeCount(input.inputTokens) +
|
||||
normalizeCount(input.outputTokens) +
|
||||
normalizeCount(input.cacheReadTokens) +
|
||||
normalizeCount(input.cacheWriteTokens) > 0;
|
||||
}
|
||||
|
||||
async function getPriceCatalog(): Promise<PriceCatalog> {
|
||||
if (catalog && Date.now() - catalog.loadedAt < catalogTtlMs) {
|
||||
return catalog;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,418 @@
|
||||
import { existsSync, mkdirSync } from "node:fs";
|
||||
import { dirname } from "node:path";
|
||||
import {
|
||||
createBetterSqliteDatabase,
|
||||
type BetterSqliteDatabase,
|
||||
type BetterSqliteStatement
|
||||
} from "@ccr/core/storage/sqlite-native";
|
||||
|
||||
export type RequestLogAdmission = {
|
||||
accepted: boolean;
|
||||
bodyCapturePolicy: "all" | "errors" | "none";
|
||||
bodyCaptureMaxBytes: number;
|
||||
reason?: string;
|
||||
recordedAt: number;
|
||||
state: "committed" | "pending" | "rejected";
|
||||
};
|
||||
|
||||
export type RawTraceAdmissionResolution = RequestLogAdmission | "pending";
|
||||
|
||||
const runtimeLeaseMs = 30_000;
|
||||
const terminalAdmissionRetentionMs = 48 * 60 * 60 * 1_000;
|
||||
const requestLogsReconnectCooldownMs = 1_000;
|
||||
|
||||
export class RequestLogAdmissionStore {
|
||||
private readonly database: BetterSqliteDatabase;
|
||||
private readonly pendingInsertStatement: BetterSqliteStatement;
|
||||
private readonly pendingReadStatement: BetterSqliteStatement;
|
||||
private readonly readStatement: BetterSqliteStatement;
|
||||
private requestLogsDatabase?: BetterSqliteDatabase;
|
||||
private readonly requestLogsDbFile: string;
|
||||
private requestLogsReadStatement?: BetterSqliteStatement;
|
||||
private requestLogsUnavailableUntil = 0;
|
||||
private readonly runtimeId: string;
|
||||
private readonly upsertStatement: BetterSqliteStatement;
|
||||
|
||||
constructor(dbFile: string, requestLogsDbFile: string, runtimeId: string) {
|
||||
mkdirSync(dirname(dbFile), { recursive: true });
|
||||
const database = createBetterSqliteDatabase(dbFile);
|
||||
try {
|
||||
database.pragma("journal_mode = WAL");
|
||||
database.pragma("synchronous = NORMAL");
|
||||
// Admission operations run on the gateway thread. Fail immediately on a
|
||||
// competing writer; RequestLogRuntime retains and retries the exact
|
||||
// operation asynchronously without blocking the event loop.
|
||||
database.pragma("busy_timeout = 0");
|
||||
database.exec(`
|
||||
CREATE TABLE IF NOT EXISTS request_log_admissions (
|
||||
request_id TEXT PRIMARY KEY,
|
||||
state TEXT NOT NULL,
|
||||
runtime_id TEXT NOT NULL,
|
||||
body_capture_policy TEXT NOT NULL DEFAULT 'all',
|
||||
body_capture_max_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
reason TEXT NOT NULL DEFAULT '',
|
||||
recorded_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS request_log_admissions_state_runtime_idx
|
||||
ON request_log_admissions(state, runtime_id);
|
||||
CREATE INDEX IF NOT EXISTS request_log_admissions_recorded_at_idx
|
||||
ON request_log_admissions(recorded_at);
|
||||
CREATE TABLE IF NOT EXISTS request_log_admission_runtimes (
|
||||
runtime_id TEXT PRIMARY KEY,
|
||||
owner_pid INTEGER NOT NULL,
|
||||
heartbeat_at INTEGER NOT NULL,
|
||||
lease_expires_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS request_log_admission_runtimes_lease_idx
|
||||
ON request_log_admission_runtimes(lease_expires_at);
|
||||
CREATE TABLE IF NOT EXISTS request_log_raw_admission_pending (
|
||||
request_id TEXT PRIMARY KEY,
|
||||
first_seen_at INTEGER NOT NULL,
|
||||
last_seen_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS request_log_raw_admission_pending_seen_idx
|
||||
ON request_log_raw_admission_pending(first_seen_at);
|
||||
`);
|
||||
ensureAdmissionBodyCapturePolicyColumn(database);
|
||||
// Older builds marked the whole logical request consumed after the first
|
||||
// raw bundle. A logical request can own several fallback bundles.
|
||||
database.exec(`
|
||||
UPDATE request_log_admissions
|
||||
SET state = 'committed', reason = ''
|
||||
WHERE state = 'consumed'
|
||||
`);
|
||||
this.database = database;
|
||||
this.requestLogsDbFile = requestLogsDbFile;
|
||||
this.runtimeId = runtimeId;
|
||||
this.readStatement = database.prepare(`
|
||||
SELECT state, body_capture_policy, body_capture_max_bytes, reason, recorded_at
|
||||
FROM request_log_admissions
|
||||
WHERE request_id = ?
|
||||
LIMIT 1
|
||||
`);
|
||||
this.upsertStatement = database.prepare(`
|
||||
INSERT INTO request_log_admissions (
|
||||
request_id,
|
||||
state,
|
||||
runtime_id,
|
||||
body_capture_policy,
|
||||
body_capture_max_bytes,
|
||||
reason,
|
||||
recorded_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(request_id) DO UPDATE SET
|
||||
state = excluded.state,
|
||||
runtime_id = excluded.runtime_id,
|
||||
body_capture_policy = excluded.body_capture_policy,
|
||||
body_capture_max_bytes = excluded.body_capture_max_bytes,
|
||||
reason = excluded.reason,
|
||||
recorded_at = excluded.recorded_at
|
||||
`);
|
||||
this.pendingReadStatement = database.prepare(`
|
||||
SELECT first_seen_at
|
||||
FROM request_log_raw_admission_pending
|
||||
WHERE request_id = ?
|
||||
`);
|
||||
this.pendingInsertStatement = database.prepare(`
|
||||
INSERT INTO request_log_raw_admission_pending (request_id, first_seen_at, last_seen_at)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(request_id) DO NOTHING
|
||||
`);
|
||||
this.heartbeat(runtimeId);
|
||||
this.prune();
|
||||
} catch (error) {
|
||||
database.close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
close(): void {
|
||||
try {
|
||||
this.database.prepare("DELETE FROM request_log_admission_runtimes WHERE runtime_id = ?")
|
||||
.run(this.runtimeId);
|
||||
} finally {
|
||||
this.closeRequestLogsDatabase();
|
||||
this.database.close();
|
||||
}
|
||||
}
|
||||
|
||||
heartbeat(runtimeId: string): void {
|
||||
const now = Date.now();
|
||||
this.database.prepare(`
|
||||
INSERT INTO request_log_admission_runtimes (
|
||||
runtime_id,
|
||||
owner_pid,
|
||||
heartbeat_at,
|
||||
lease_expires_at
|
||||
) VALUES (?, ?, ?, ?)
|
||||
ON CONFLICT(runtime_id) DO UPDATE SET
|
||||
owner_pid = excluded.owner_pid,
|
||||
heartbeat_at = excluded.heartbeat_at,
|
||||
lease_expires_at = excluded.lease_expires_at
|
||||
`).run(runtimeId, process.pid, now, now + runtimeLeaseMs);
|
||||
reconcileInterruptedAdmissions(this.database, this.requestLogsDbFile, runtimeId);
|
||||
}
|
||||
|
||||
markCommitted(requestId: string, runtimeId: string): void {
|
||||
this.database.prepare(`
|
||||
UPDATE request_log_admissions
|
||||
SET state = 'committed', reason = '', recorded_at = ?
|
||||
WHERE request_id = ? AND state = 'pending' AND runtime_id = ?
|
||||
`).run(Date.now(), requestId, runtimeId);
|
||||
}
|
||||
|
||||
prune(now = Date.now()): void {
|
||||
this.database.prepare(`
|
||||
DELETE FROM request_log_admissions
|
||||
WHERE state IN ('committed', 'rejected') AND recorded_at < ?
|
||||
`).run(now - terminalAdmissionRetentionMs);
|
||||
this.database.prepare(`
|
||||
DELETE FROM request_log_raw_admission_pending
|
||||
WHERE first_seen_at < ?
|
||||
`).run(now - terminalAdmissionRetentionMs);
|
||||
|
||||
this.database.prepare(`
|
||||
DELETE FROM request_log_admission_runtimes
|
||||
WHERE lease_expires_at < ? AND runtime_id <> ?
|
||||
`).run(now - terminalAdmissionRetentionMs, this.runtimeId);
|
||||
}
|
||||
|
||||
read(requestId: string): RequestLogAdmission | undefined {
|
||||
const row = this.readStatement.get(requestId) as Record<string, unknown> | undefined;
|
||||
return row ? admissionFromRow(row) : undefined;
|
||||
}
|
||||
|
||||
remember(input: {
|
||||
accepted: boolean;
|
||||
bodyCapturePolicy?: "all" | "errors" | "none";
|
||||
bodyCaptureMaxBytes: number;
|
||||
reason?: string;
|
||||
requestId: string;
|
||||
runtimeId: string;
|
||||
}): void {
|
||||
this.database.exec("BEGIN IMMEDIATE");
|
||||
try {
|
||||
this.upsertStatement.run(
|
||||
input.requestId,
|
||||
input.accepted ? "pending" : "rejected",
|
||||
input.runtimeId,
|
||||
input.bodyCapturePolicy ?? "all",
|
||||
nonNegativeInteger(input.bodyCaptureMaxBytes),
|
||||
input.reason ?? "",
|
||||
Date.now()
|
||||
);
|
||||
this.database.prepare("DELETE FROM request_log_raw_admission_pending WHERE request_id = ?")
|
||||
.run(input.requestId);
|
||||
this.database.exec("COMMIT");
|
||||
} catch (error) {
|
||||
if (this.database.inTransaction) this.database.exec("ROLLBACK");
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
resolveForRawTrace(requestId: string, pendingTtlMs: number): RawTraceAdmissionResolution {
|
||||
const existing = this.read(requestId);
|
||||
if (existing && existing.state !== "pending") return existing;
|
||||
|
||||
const committed = this.readCommittedRequestLogAdmission(requestId);
|
||||
if (committed.state === "found") return committed.admission;
|
||||
if (committed.state === "unavailable") return "pending";
|
||||
if (existing) return existing;
|
||||
|
||||
const now = Date.now();
|
||||
const pending = this.pendingReadStatement.get(requestId) as Record<string, unknown> | undefined;
|
||||
const firstSeenAt = pending ? nonNegativeInteger(pending.first_seen_at) : now;
|
||||
if (pending && now - firstSeenAt < Math.max(1, pendingTtlMs)) {
|
||||
// Polling an already-pending admission is read-only. In particular, do
|
||||
// not update last_seen_at on every raw-trace replay.
|
||||
return "pending";
|
||||
}
|
||||
|
||||
this.database.exec("BEGIN IMMEDIATE");
|
||||
try {
|
||||
const racedAdmission = this.read(requestId);
|
||||
if (racedAdmission) {
|
||||
this.database.exec("COMMIT");
|
||||
return racedAdmission;
|
||||
}
|
||||
const racedPending = this.pendingReadStatement.get(requestId) as Record<string, unknown> | undefined;
|
||||
const racedFirstSeenAt = racedPending ? nonNegativeInteger(racedPending.first_seen_at) : firstSeenAt;
|
||||
if (racedPending && now - racedFirstSeenAt >= Math.max(1, pendingTtlMs)) {
|
||||
this.upsertStatement.run(requestId, "rejected", this.runtimeId, "none", 0, "record_missing", now);
|
||||
this.database.prepare("DELETE FROM request_log_raw_admission_pending WHERE request_id = ?")
|
||||
.run(requestId);
|
||||
this.database.exec("COMMIT");
|
||||
return this.read(requestId)!;
|
||||
}
|
||||
this.pendingInsertStatement.run(requestId, racedFirstSeenAt, now);
|
||||
this.database.exec("COMMIT");
|
||||
return "pending";
|
||||
} catch (error) {
|
||||
if (this.database.inTransaction) this.database.exec("ROLLBACK");
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private closeRequestLogsDatabase(): void {
|
||||
try {
|
||||
this.requestLogsDatabase?.close();
|
||||
} catch {
|
||||
// Closing the primary admission database must still be attempted.
|
||||
}
|
||||
this.requestLogsDatabase = undefined;
|
||||
this.requestLogsReadStatement = undefined;
|
||||
}
|
||||
|
||||
private readCommittedRequestLogAdmission(requestId: string): CommittedAdmissionLookup {
|
||||
if (!existsSync(this.requestLogsDbFile)) return { state: "missing" };
|
||||
if (Date.now() < this.requestLogsUnavailableUntil) return { state: "unavailable" };
|
||||
try {
|
||||
if (!this.requestLogsDatabase) {
|
||||
this.requestLogsDatabase = createBetterSqliteDatabase(this.requestLogsDbFile, {
|
||||
fileMustExist: true,
|
||||
readonly: true
|
||||
});
|
||||
this.requestLogsDatabase.pragma("busy_timeout = 0");
|
||||
this.requestLogsReadStatement = this.requestLogsDatabase.prepare(`
|
||||
SELECT
|
||||
gateway_body_capture_policy,
|
||||
gateway_body_capture_max_bytes,
|
||||
completed_at
|
||||
FROM request_logs
|
||||
WHERE request_id = ?
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
`);
|
||||
}
|
||||
this.requestLogsUnavailableUntil = 0;
|
||||
const row = this.requestLogsReadStatement!.get(requestId) as Record<string, unknown> | undefined;
|
||||
if (!row) return { state: "missing" };
|
||||
return {
|
||||
admission: {
|
||||
accepted: true,
|
||||
bodyCapturePolicy: normalizeBodyCapturePolicy(row.gateway_body_capture_policy),
|
||||
bodyCaptureMaxBytes: nonNegativeInteger(row.gateway_body_capture_max_bytes),
|
||||
recordedAt: dateMs(row.completed_at),
|
||||
state: "committed"
|
||||
},
|
||||
state: "found"
|
||||
};
|
||||
} catch {
|
||||
this.closeRequestLogsDatabase();
|
||||
this.requestLogsUnavailableUntil = Date.now() + requestLogsReconnectCooldownMs;
|
||||
return { state: "unavailable" };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function reconcileInterruptedAdmissions(
|
||||
database: BetterSqliteDatabase,
|
||||
requestLogsDbFile: string,
|
||||
runtimeId: string
|
||||
): void {
|
||||
const now = Date.now();
|
||||
const owners = database.prepare(`
|
||||
SELECT
|
||||
admissions.runtime_id,
|
||||
runtimes.lease_expires_at
|
||||
FROM request_log_admissions AS admissions
|
||||
LEFT JOIN request_log_admission_runtimes AS runtimes
|
||||
ON runtimes.runtime_id = admissions.runtime_id
|
||||
WHERE admissions.state = 'pending' AND admissions.runtime_id <> ?
|
||||
GROUP BY admissions.runtime_id
|
||||
`).all(runtimeId) as Array<Record<string, unknown>>;
|
||||
const interruptedRuntimeIds = owners.filter((owner) => {
|
||||
const leaseExpiresAt = nonNegativeInteger(owner.lease_expires_at);
|
||||
return leaseExpiresAt <= now;
|
||||
}).map((owner) => String(owner.runtime_id ?? "")).filter(Boolean);
|
||||
if (interruptedRuntimeIds.length === 0) return;
|
||||
|
||||
let attached = false;
|
||||
try {
|
||||
database.prepare("ATTACH DATABASE ? AS request_logs_db").run(requestLogsDbFile);
|
||||
attached = true;
|
||||
for (const interruptedRuntimeId of interruptedRuntimeIds) {
|
||||
database.prepare(`
|
||||
UPDATE request_log_admissions
|
||||
SET
|
||||
state = CASE
|
||||
WHEN EXISTS (
|
||||
SELECT 1
|
||||
FROM request_logs_db.request_logs
|
||||
WHERE request_logs.request_id = request_log_admissions.request_id
|
||||
) THEN 'committed'
|
||||
ELSE 'rejected'
|
||||
END,
|
||||
reason = CASE
|
||||
WHEN EXISTS (
|
||||
SELECT 1
|
||||
FROM request_logs_db.request_logs
|
||||
WHERE request_logs.request_id = request_log_admissions.request_id
|
||||
) THEN ''
|
||||
ELSE 'writer_unavailable'
|
||||
END,
|
||||
recorded_at = ?
|
||||
WHERE state = 'pending' AND runtime_id = ?
|
||||
`).run(now, interruptedRuntimeId);
|
||||
}
|
||||
} catch (error) {
|
||||
// ATTACH/query/I/O failures do not prove that the request log is absent.
|
||||
// Keep admissions pending so the next heartbeat can reconcile them.
|
||||
console.warn(`[request-log] Admission reconciliation deferred: ${formatError(error)}`);
|
||||
} finally {
|
||||
if (attached) database.exec("DETACH DATABASE request_logs_db");
|
||||
}
|
||||
}
|
||||
|
||||
type CommittedAdmissionLookup = {
|
||||
admission: RequestLogAdmission;
|
||||
state: "found";
|
||||
} | {
|
||||
state: "missing" | "unavailable";
|
||||
};
|
||||
|
||||
function admissionFromRow(row: Record<string, unknown>): RequestLogAdmission {
|
||||
const state = normalizeState(row.state);
|
||||
return {
|
||||
accepted: state === "committed" || state === "pending",
|
||||
bodyCapturePolicy: normalizeBodyCapturePolicy(row.body_capture_policy),
|
||||
bodyCaptureMaxBytes: nonNegativeInteger(row.body_capture_max_bytes),
|
||||
reason: stringValue(row.reason),
|
||||
recordedAt: nonNegativeInteger(row.recorded_at),
|
||||
state
|
||||
};
|
||||
}
|
||||
|
||||
function ensureAdmissionBodyCapturePolicyColumn(database: BetterSqliteDatabase): void {
|
||||
const columns = database.prepare("PRAGMA table_info('request_log_admissions')").all() as Array<Record<string, unknown>>;
|
||||
if (!columns.some((column) => column.name === "body_capture_policy")) {
|
||||
database.exec("ALTER TABLE request_log_admissions ADD COLUMN body_capture_policy TEXT NOT NULL DEFAULT 'all'");
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeBodyCapturePolicy(value: unknown): RequestLogAdmission["bodyCapturePolicy"] {
|
||||
return value === "errors" || value === "none" ? value : "all";
|
||||
}
|
||||
|
||||
function normalizeState(value: unknown): RequestLogAdmission["state"] {
|
||||
if (value === "committed" || value === "pending") return value;
|
||||
return "rejected";
|
||||
}
|
||||
|
||||
function nonNegativeInteger(value: unknown): number {
|
||||
const parsed = Number(value);
|
||||
return Number.isFinite(parsed) ? Math.max(0, Math.floor(parsed)) : 0;
|
||||
}
|
||||
|
||||
function stringValue(value: unknown): string | undefined {
|
||||
return typeof value === "string" && value ? value : undefined;
|
||||
}
|
||||
|
||||
function dateMs(value: unknown): number {
|
||||
const parsed = Date.parse(String(value ?? ""));
|
||||
return Number.isFinite(parsed) ? parsed : Date.now();
|
||||
}
|
||||
|
||||
function formatError(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
@@ -26,15 +26,23 @@ import {
|
||||
resolveRawTraceBodyLimit
|
||||
} from "@ccr/core/observability/request-log-limits";
|
||||
import { compactBase64ImagePayloads } from "@ccr/core/observability/request-log-body";
|
||||
import {
|
||||
RequestLogAdmissionStore,
|
||||
type RequestLogAdmission
|
||||
} from "@ccr/core/observability/request-log-admission-store";
|
||||
import { suppressRouteTraceBodyValues } from "@ccr/core/observability/route-trace";
|
||||
import { isSensitiveRequestLogHeaderName } from "@ccr/core/observability/sensitive-headers";
|
||||
|
||||
export type RequestLogEnqueueResult = {
|
||||
accepted: boolean;
|
||||
degraded: boolean;
|
||||
reason?: "body_removed" | "closed" | "queue_full" | "writer_unavailable";
|
||||
reason?: "body_removed" | "closed" | "queue_full" | "record_dropped" | "record_pending" | "writer_unavailable";
|
||||
};
|
||||
|
||||
export type RequestLogRuntimeMetrics = {
|
||||
accepted: number;
|
||||
admissionOverlayItems: number;
|
||||
admissionPendingOperations: number;
|
||||
committed: number;
|
||||
degraded: number;
|
||||
dropped: number;
|
||||
@@ -45,24 +53,33 @@ export type RequestLogRuntimeMetrics = {
|
||||
};
|
||||
|
||||
export type RequestLogRuntimeOptions = {
|
||||
admissionDbFile?: string;
|
||||
admissionMaxPendingOperations?: number;
|
||||
admissionOperationMaxAgeMs?: number;
|
||||
admissionOverlayMaxEntries?: number;
|
||||
admissionOverlayTtlMs?: number;
|
||||
batchMaxBytes?: number;
|
||||
batchMaxItems?: number;
|
||||
batchMaxWaitMs?: number;
|
||||
dbFile: string;
|
||||
queueMaxBytes?: number;
|
||||
queueMaxItems?: number;
|
||||
pendingAdmissionTtlMs?: number;
|
||||
rawTraceSpoolDir?: string;
|
||||
workerFile?: string;
|
||||
};
|
||||
|
||||
type ResolvedRuntimeOptions = Required<Omit<RequestLogRuntimeOptions, "rawTraceSpoolDir" | "workerFile">> & {
|
||||
type ResolvedRuntimeOptions = Required<Omit<RequestLogRuntimeOptions, "admissionDbFile" | "rawTraceSpoolDir" | "workerFile">> & {
|
||||
admissionDbFile: string;
|
||||
rawTraceSpoolDir: string;
|
||||
workerFile: string;
|
||||
};
|
||||
|
||||
type QueuedCommand = RequestLogStoreWriteCommand & {
|
||||
batchBytes: number;
|
||||
isolated?: boolean;
|
||||
sizeBytes: number;
|
||||
writeAttempts: number;
|
||||
};
|
||||
|
||||
type InFlightBatch = {
|
||||
@@ -75,7 +92,8 @@ type WorkerResponse = {
|
||||
error?: string;
|
||||
requestId?: number;
|
||||
result?: unknown;
|
||||
type: "ack" | "batch-error" | "ready" | "response";
|
||||
type: "ack" | "batch-error" | "maintenance" | "ready" | "response";
|
||||
updated?: number;
|
||||
};
|
||||
|
||||
type PendingRpc = {
|
||||
@@ -83,22 +101,42 @@ type PendingRpc = {
|
||||
resolve: (value: unknown) => void;
|
||||
};
|
||||
|
||||
type AdmissionOperation = {
|
||||
attempts: number;
|
||||
createdAt: number;
|
||||
key?: string;
|
||||
overlayRequestId?: string;
|
||||
overlayVersion?: number;
|
||||
run: (store: RequestLogAdmissionStore) => void;
|
||||
};
|
||||
|
||||
type AdmissionOverlayEntry = RequestLogAdmission & {
|
||||
version: number;
|
||||
};
|
||||
|
||||
// One raw-trace event can contain both a maximum-sized request and response.
|
||||
// Keep room for exactly that case while retaining byte-based backpressure for
|
||||
// concurrent events.
|
||||
const defaultQueueMaxBytes = 128 * 1024 * 1024;
|
||||
const sensitiveQueueHeaderNames = new Set([
|
||||
"authorization",
|
||||
"cookie",
|
||||
"proxy-authorization",
|
||||
"set-cookie",
|
||||
"x-api-key",
|
||||
"x-auth-api-key-id",
|
||||
"x-auth-sub"
|
||||
]);
|
||||
const maxCommandWriteAttempts = 3;
|
||||
const admissionRetryMaxDelayMs = 5_000;
|
||||
const defaultAdmissionMaxPendingOperations = 20_000;
|
||||
const defaultAdmissionOperationMaxAgeMs = 10 * 60 * 1_000;
|
||||
const defaultAdmissionOverlayMaxEntries = 10_000;
|
||||
const defaultAdmissionOverlayTtlMs = 10 * 60 * 1_000;
|
||||
const admissionDrainTimeSliceMs = 5;
|
||||
const admissionDrainMaxOperations = 100;
|
||||
|
||||
export class RequestLogRuntime {
|
||||
private accepted = 0;
|
||||
private admissionHeartbeatTimer?: NodeJS.Timeout;
|
||||
private admissionLastPrunedAt = 0;
|
||||
private admissionLastWarningAt = 0;
|
||||
private readonly admissionOperationKeys = new Set<string>();
|
||||
private readonly admissionOperations = new Map<number, AdmissionOperation>();
|
||||
private readonly admissionOverlay = new Map<string, AdmissionOverlayEntry>();
|
||||
private admissionRetryTimer?: NodeJS.Timeout;
|
||||
private admissionStore?: RequestLogAdmissionStore;
|
||||
private batchId = 0;
|
||||
private closed = false;
|
||||
private committed = 0;
|
||||
@@ -108,6 +146,8 @@ export class RequestLogRuntime {
|
||||
private flushTimer?: NodeJS.Timeout;
|
||||
private readonly inFlight = new Map<number, InFlightBatch>();
|
||||
private nextRequestId = 0;
|
||||
private nextAdmissionOperationId = 0;
|
||||
private nextAdmissionOverlayVersion = 0;
|
||||
private nextSequence = 0;
|
||||
private readonly options: ResolvedRuntimeOptions;
|
||||
private outstandingBytes = 0;
|
||||
@@ -115,6 +155,8 @@ export class RequestLogRuntime {
|
||||
private queryWorker?: Worker;
|
||||
private queryWorkerReady?: Promise<void>;
|
||||
private readonly queue: QueuedCommand[] = [];
|
||||
private revision = 0;
|
||||
private readonly runtimeId = randomUUID();
|
||||
private readonly writerRequests = new Map<number, PendingRpc>();
|
||||
private writerRestartCount = 0;
|
||||
private writerWorker?: Worker;
|
||||
@@ -122,10 +164,28 @@ export class RequestLogRuntime {
|
||||
|
||||
constructor(options: RequestLogRuntimeOptions) {
|
||||
this.options = {
|
||||
admissionDbFile: options.admissionDbFile ?? `${options.dbFile}.admissions.sqlite`,
|
||||
admissionMaxPendingOperations: positiveInteger(
|
||||
options.admissionMaxPendingOperations,
|
||||
defaultAdmissionMaxPendingOperations
|
||||
),
|
||||
admissionOperationMaxAgeMs: positiveInteger(
|
||||
options.admissionOperationMaxAgeMs,
|
||||
defaultAdmissionOperationMaxAgeMs
|
||||
),
|
||||
admissionOverlayMaxEntries: positiveInteger(
|
||||
options.admissionOverlayMaxEntries,
|
||||
defaultAdmissionOverlayMaxEntries
|
||||
),
|
||||
admissionOverlayTtlMs: positiveInteger(
|
||||
options.admissionOverlayTtlMs,
|
||||
defaultAdmissionOverlayTtlMs
|
||||
),
|
||||
batchMaxBytes: positiveInteger(options.batchMaxBytes, 4 * 1024 * 1024),
|
||||
batchMaxItems: positiveInteger(options.batchMaxItems, 50),
|
||||
batchMaxWaitMs: positiveInteger(options.batchMaxWaitMs, 10),
|
||||
dbFile: options.dbFile,
|
||||
pendingAdmissionTtlMs: positiveInteger(options.pendingAdmissionTtlMs, 5 * 60 * 1_000),
|
||||
queueMaxBytes: positiveInteger(options.queueMaxBytes, defaultQueueMaxBytes),
|
||||
queueMaxItems: positiveInteger(options.queueMaxItems, 2_000),
|
||||
rawTraceSpoolDir: options.rawTraceSpoolDir ?? RAW_TRACE_SPOOL_DIR,
|
||||
@@ -138,7 +198,18 @@ export class RequestLogRuntime {
|
||||
const ordinarySuccess = input.statusCode >= 200 && input.statusCode < 400 && !input.error;
|
||||
if (pressure >= 0.95 && ordinarySuccess) {
|
||||
this.dropped += 1;
|
||||
return { accepted: false, degraded: false, reason: "queue_full" };
|
||||
const result: RequestLogEnqueueResult = {
|
||||
accepted: false,
|
||||
degraded: false,
|
||||
reason: "queue_full"
|
||||
};
|
||||
this.rememberRecordAdmission(
|
||||
input.requestId,
|
||||
result,
|
||||
0,
|
||||
resolveRecordBodyCapturePolicy(input)
|
||||
);
|
||||
return result;
|
||||
}
|
||||
const prepared = prepareRecordForQueue(input, pressure);
|
||||
const sizeBytes = estimateRecordBytes(prepared.input);
|
||||
@@ -148,32 +219,86 @@ export class RequestLogRuntime {
|
||||
kind: "record",
|
||||
sequence: ++this.nextSequence,
|
||||
batchBytes: sizeBytes,
|
||||
sizeBytes
|
||||
sizeBytes,
|
||||
writeAttempts: 0
|
||||
};
|
||||
const result = this.enqueue(command, prepared.degraded ? "body_removed" : undefined);
|
||||
this.rememberRecordAdmission(
|
||||
input.requestId,
|
||||
result,
|
||||
prepared.bodyCaptureMaxBytes,
|
||||
resolveRecordBodyCapturePolicy(input)
|
||||
);
|
||||
if (result.accepted && prepared.degraded) this.degraded += 1;
|
||||
return result;
|
||||
}
|
||||
|
||||
rejectRecord(requestId: string | undefined, reason = "sampled"): void {
|
||||
this.rememberRecordAdmission(requestId, {
|
||||
accepted: false,
|
||||
degraded: false,
|
||||
reason: "record_dropped"
|
||||
}, 0, "none", reason);
|
||||
}
|
||||
|
||||
enqueueRawTrace(
|
||||
input: RequestLogRawTraceUpdateInput,
|
||||
rawTraceFiles?: RequestLogRawTraceFiles
|
||||
): RequestLogEnqueueResult {
|
||||
const maxBodyBytes = resolveRawTraceBodyLimit(rawTraceFiles?.maxBodyBytes);
|
||||
const prepared = prepareRawTraceForQueue(input, maxBodyBytes);
|
||||
const recordAdmission = input.deferOutcomeUntilRecord
|
||||
? this.resolveRecordAdmission(input.requestId)
|
||||
: this.readRecordAdmission(input.requestId);
|
||||
if (recordAdmission === "pending" ||
|
||||
(input.deferOutcomeUntilRecord && recordAdmission?.state === "pending") ||
|
||||
(recordAdmission === undefined && input.deferOutcomeUntilRecord)) {
|
||||
return {
|
||||
accepted: false,
|
||||
degraded: false,
|
||||
reason: "record_pending"
|
||||
};
|
||||
}
|
||||
if (recordAdmission && !recordAdmission.accepted) {
|
||||
this.dropped += 1;
|
||||
return {
|
||||
accepted: false,
|
||||
degraded: false,
|
||||
reason: "record_dropped"
|
||||
};
|
||||
}
|
||||
const ordinarySuccess = input.statusCode !== undefined &&
|
||||
input.statusCode >= 200 && input.statusCode < 400;
|
||||
if (!recordAdmission && this.pressureRatio() >= 0.95 && ordinarySuccess) {
|
||||
this.dropped += 1;
|
||||
return { accepted: false, degraded: false, reason: "queue_full" };
|
||||
}
|
||||
const configuredMaxBodyBytes = resolveRawTraceBodyLimit(rawTraceFiles?.maxBodyBytes);
|
||||
const maxBodyBytes = Math.min(
|
||||
configuredMaxBodyBytes,
|
||||
recordAdmission?.bodyCaptureMaxBytes ?? configuredMaxBodyBytes
|
||||
);
|
||||
const bodyPolicyDegraded = maxBodyBytes < configuredMaxBodyBytes;
|
||||
const admittedInput = recordAdmission
|
||||
? { ...input, bodyCapturePolicy: recordAdmission.bodyCapturePolicy }
|
||||
: input;
|
||||
const policyInput = maxBodyBytes === 0
|
||||
? suppressRequestLogRawTraceBodies(admittedInput)
|
||||
: admittedInput;
|
||||
const queuedRawTraceFiles = constrainRawTraceFiles(rawTraceFiles, maxBodyBytes);
|
||||
const prepared = prepareRawTraceForQueue(policyInput, maxBodyBytes);
|
||||
const sizeBytes = Math.max(
|
||||
estimateRawTraceBytes(prepared, rawTraceFiles),
|
||||
rawTraceFileBytes(rawTraceFiles, maxBodyBytes)
|
||||
estimateRawTraceBytes(prepared, queuedRawTraceFiles),
|
||||
rawTraceFileBytes(queuedRawTraceFiles, maxBodyBytes)
|
||||
);
|
||||
const command: QueuedCommand = {
|
||||
input: prepared,
|
||||
kind: "raw-trace-update",
|
||||
rawTraceFiles,
|
||||
rawTraceFiles: queuedRawTraceFiles,
|
||||
sequence: ++this.nextSequence,
|
||||
batchBytes: sizeBytes,
|
||||
sizeBytes
|
||||
sizeBytes,
|
||||
writeAttempts: 0
|
||||
};
|
||||
return this.enqueue(command);
|
||||
return this.enqueue(command, bodyPolicyDegraded ? "body_removed" : undefined);
|
||||
}
|
||||
|
||||
async list(filter?: RequestLogListFilter): Promise<RequestLogPage> {
|
||||
@@ -193,8 +318,11 @@ export class RequestLogRuntime {
|
||||
}
|
||||
|
||||
metrics(): RequestLogRuntimeMetrics {
|
||||
this.pruneAdmissionOverlay();
|
||||
return {
|
||||
accepted: this.accepted,
|
||||
admissionOverlayItems: this.admissionOverlay.size,
|
||||
admissionPendingOperations: this.admissionOperations.size,
|
||||
committed: this.committed,
|
||||
degraded: this.degraded,
|
||||
dropped: this.dropped,
|
||||
@@ -234,6 +362,21 @@ export class RequestLogRuntime {
|
||||
]);
|
||||
this.queryWorker = undefined;
|
||||
this.writerWorker = undefined;
|
||||
if (this.admissionHeartbeatTimer) clearInterval(this.admissionHeartbeatTimer);
|
||||
this.admissionHeartbeatTimer = undefined;
|
||||
this.drainAdmissionOperations();
|
||||
await waitUntil(() => this.admissionOperations.size === 0, options.timeoutMs);
|
||||
if (this.admissionRetryTimer) clearTimeout(this.admissionRetryTimer);
|
||||
this.admissionRetryTimer = undefined;
|
||||
this.admissionOperations.clear();
|
||||
this.admissionOperationKeys.clear();
|
||||
this.admissionOverlay.clear();
|
||||
try {
|
||||
this.admissionStore?.close();
|
||||
} catch (error) {
|
||||
console.warn(`[request-log] Failed to close admission persistence: ${formatRuntimeError(error)}`);
|
||||
}
|
||||
this.admissionStore = undefined;
|
||||
}
|
||||
|
||||
private enqueue(command: QueuedCommand, degradedReason?: RequestLogEnqueueResult["reason"]): RequestLogEnqueueResult {
|
||||
@@ -274,6 +417,239 @@ export class RequestLogRuntime {
|
||||
return this.queue.reduce((total, command) => total + command.batchBytes, 0);
|
||||
}
|
||||
|
||||
private rememberRecordAdmission(
|
||||
requestId: string | undefined,
|
||||
result: RequestLogEnqueueResult,
|
||||
bodyCaptureMaxBytes: number,
|
||||
bodyCapturePolicy: "all" | "errors" | "none",
|
||||
persistedReason: string | undefined = result.reason
|
||||
): void {
|
||||
const normalized = requestId?.trim();
|
||||
if (!normalized) return;
|
||||
const overlayVersion = this.setAdmissionOverlay(normalized, {
|
||||
accepted: result.accepted,
|
||||
bodyCapturePolicy,
|
||||
bodyCaptureMaxBytes,
|
||||
reason: persistedReason,
|
||||
recordedAt: Date.now(),
|
||||
state: result.accepted ? "pending" : "rejected"
|
||||
});
|
||||
this.submitAdmissionOperation({
|
||||
attempts: 0,
|
||||
createdAt: Date.now(),
|
||||
overlayRequestId: normalized,
|
||||
overlayVersion,
|
||||
run: (store) => store.remember({
|
||||
accepted: result.accepted,
|
||||
bodyCapturePolicy,
|
||||
bodyCaptureMaxBytes,
|
||||
reason: persistedReason,
|
||||
requestId: normalized,
|
||||
runtimeId: this.runtimeId
|
||||
})
|
||||
});
|
||||
}
|
||||
|
||||
private readRecordAdmission(requestId: string): RequestLogAdmission | undefined {
|
||||
const normalized = requestId.trim();
|
||||
if (!normalized) return undefined;
|
||||
this.pruneAdmissionOverlay();
|
||||
const overlay = this.admissionOverlay.get(normalized);
|
||||
if (overlay) return overlay;
|
||||
return this.useAdmissionStore((store) => store.read(normalized));
|
||||
}
|
||||
|
||||
private resolveRecordAdmission(requestId: string): RequestLogAdmission | "pending" | undefined {
|
||||
const normalized = requestId.trim();
|
||||
if (!normalized) return undefined;
|
||||
this.pruneAdmissionOverlay();
|
||||
const overlay = this.admissionOverlay.get(normalized);
|
||||
if (overlay) return overlay;
|
||||
return this.useAdmissionStore((store) =>
|
||||
store.resolveForRawTrace(normalized, this.options.pendingAdmissionTtlMs));
|
||||
}
|
||||
|
||||
private markRecordAdmissionsCommitted(commands: QueuedCommand[]): void {
|
||||
for (const command of commands) {
|
||||
if (command.kind !== "record") continue;
|
||||
const requestId = command.input.requestId?.trim();
|
||||
if (!requestId) continue;
|
||||
const existing = this.admissionOverlay.get(requestId);
|
||||
const overlayVersion = this.setAdmissionOverlay(requestId, {
|
||||
accepted: true,
|
||||
bodyCaptureMaxBytes: existing?.bodyCaptureMaxBytes ?? nonNegativeInteger(command.input.maxBodyBytes),
|
||||
bodyCapturePolicy: existing?.bodyCapturePolicy ?? resolveRecordBodyCapturePolicy(command.input),
|
||||
recordedAt: Date.now(),
|
||||
state: "committed"
|
||||
});
|
||||
this.submitAdmissionOperation({
|
||||
attempts: 0,
|
||||
createdAt: Date.now(),
|
||||
overlayRequestId: requestId,
|
||||
overlayVersion,
|
||||
run: (store) => store.markCommitted(requestId, this.runtimeId)
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private useAdmissionStore<T>(operation: (store: RequestLogAdmissionStore) => T): T | undefined {
|
||||
try {
|
||||
const store = this.ensureAdmissionStore();
|
||||
return operation(store);
|
||||
} catch (error) {
|
||||
this.handleAdmissionFailure(error);
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
private ensureAdmissionStore(): RequestLogAdmissionStore {
|
||||
this.admissionStore ??= new RequestLogAdmissionStore(
|
||||
this.options.admissionDbFile,
|
||||
this.options.dbFile,
|
||||
this.runtimeId
|
||||
);
|
||||
this.ensureAdmissionHeartbeat();
|
||||
return this.admissionStore;
|
||||
}
|
||||
|
||||
private submitAdmissionOperation(operation: AdmissionOperation): void {
|
||||
if (operation.key && this.admissionOperationKeys.has(operation.key)) return;
|
||||
const operationId = ++this.nextAdmissionOperationId;
|
||||
this.admissionOperations.set(operationId, operation);
|
||||
if (operation.key) this.admissionOperationKeys.add(operation.key);
|
||||
while (this.admissionOperations.size > this.options.admissionMaxPendingOperations) {
|
||||
const oldest = this.admissionOperations.entries().next().value as
|
||||
[number, AdmissionOperation] | undefined;
|
||||
if (!oldest) break;
|
||||
this.settleAdmissionOperation(oldest[0], oldest[1]);
|
||||
this.warnAdmissionBound("pending operation capacity");
|
||||
}
|
||||
this.drainAdmissionOperations();
|
||||
}
|
||||
|
||||
private drainAdmissionOperations(): void {
|
||||
if (this.admissionRetryTimer) return;
|
||||
const startedAt = Date.now();
|
||||
let processed = 0;
|
||||
while (this.admissionOperations.size > 0) {
|
||||
if (processed >= admissionDrainMaxOperations || Date.now() - startedAt >= admissionDrainTimeSliceMs) {
|
||||
this.scheduleAdmissionDrain(0);
|
||||
return;
|
||||
}
|
||||
const next = this.admissionOperations.entries().next().value as
|
||||
[number, AdmissionOperation] | undefined;
|
||||
if (!next) return;
|
||||
const [operationId, operation] = next;
|
||||
if (Date.now() - operation.createdAt >= this.options.admissionOperationMaxAgeMs) {
|
||||
this.settleAdmissionOperation(operationId, operation);
|
||||
this.warnAdmissionBound("pending operation TTL");
|
||||
processed += 1;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
operation.run(this.ensureAdmissionStore());
|
||||
this.settleAdmissionOperation(operationId, operation);
|
||||
processed += 1;
|
||||
} catch (error) {
|
||||
operation.attempts += 1;
|
||||
this.handleAdmissionFailure(error);
|
||||
const delayMs = Math.min(
|
||||
admissionRetryMaxDelayMs,
|
||||
25 * (2 ** Math.min(8, operation.attempts - 1))
|
||||
);
|
||||
this.scheduleAdmissionDrain(delayMs);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private scheduleAdmissionDrain(delayMs: number): void {
|
||||
if (this.admissionRetryTimer) return;
|
||||
this.admissionRetryTimer = setTimeout(() => {
|
||||
this.admissionRetryTimer = undefined;
|
||||
this.drainAdmissionOperations();
|
||||
}, delayMs);
|
||||
this.admissionRetryTimer.unref?.();
|
||||
}
|
||||
|
||||
private handleAdmissionFailure(error: unknown): void {
|
||||
if (!isTransientSqliteLock(error)) {
|
||||
try {
|
||||
this.admissionStore?.close();
|
||||
} catch {
|
||||
// The original persistence failure is the actionable error.
|
||||
}
|
||||
this.admissionStore = undefined;
|
||||
}
|
||||
const now = Date.now();
|
||||
if (now - this.admissionLastWarningAt >= 30_000) {
|
||||
this.admissionLastWarningAt = now;
|
||||
console.warn(`[request-log] Admission persistence operation queued for retry: ${formatRuntimeError(error)}`);
|
||||
}
|
||||
}
|
||||
|
||||
private setAdmissionOverlay(
|
||||
requestId: string,
|
||||
admission: RequestLogAdmission
|
||||
): number {
|
||||
this.pruneAdmissionOverlay(admission.recordedAt);
|
||||
const version = ++this.nextAdmissionOverlayVersion;
|
||||
this.admissionOverlay.delete(requestId);
|
||||
this.admissionOverlay.set(requestId, { ...admission, version });
|
||||
while (this.admissionOverlay.size > this.options.admissionOverlayMaxEntries) {
|
||||
const oldestRequestId = this.admissionOverlay.keys().next().value as string | undefined;
|
||||
if (!oldestRequestId) break;
|
||||
this.admissionOverlay.delete(oldestRequestId);
|
||||
this.warnAdmissionBound("overlay capacity");
|
||||
}
|
||||
return version;
|
||||
}
|
||||
|
||||
private pruneAdmissionOverlay(now = Date.now()): void {
|
||||
const cutoff = now - this.options.admissionOverlayTtlMs;
|
||||
for (const [requestId, admission] of this.admissionOverlay) {
|
||||
if (admission.recordedAt > cutoff) break;
|
||||
this.admissionOverlay.delete(requestId);
|
||||
}
|
||||
}
|
||||
|
||||
private settleAdmissionOperation(operationId: number, operation: AdmissionOperation): void {
|
||||
this.admissionOperations.delete(operationId);
|
||||
if (operation.key) this.admissionOperationKeys.delete(operation.key);
|
||||
if (!operation.overlayRequestId || operation.overlayVersion === undefined) return;
|
||||
const overlay = this.admissionOverlay.get(operation.overlayRequestId);
|
||||
if (overlay?.version === operation.overlayVersion) {
|
||||
this.admissionOverlay.delete(operation.overlayRequestId);
|
||||
}
|
||||
}
|
||||
|
||||
private warnAdmissionBound(bound: string): void {
|
||||
const now = Date.now();
|
||||
if (now - this.admissionLastWarningAt < 30_000) return;
|
||||
this.admissionLastWarningAt = now;
|
||||
console.warn(`[request-log] Admission ${bound} reached; oldest fail-closed state was released.`);
|
||||
}
|
||||
|
||||
private ensureAdmissionHeartbeat(): void {
|
||||
if (this.admissionHeartbeatTimer || this.closed) return;
|
||||
this.admissionHeartbeatTimer = setInterval(() => {
|
||||
this.submitAdmissionOperation({
|
||||
attempts: 0,
|
||||
createdAt: Date.now(),
|
||||
key: "heartbeat",
|
||||
run: (store) => {
|
||||
store.heartbeat(this.runtimeId);
|
||||
const now = Date.now();
|
||||
if (now - this.admissionLastPrunedAt >= 60 * 60 * 1_000) {
|
||||
store.prune(now);
|
||||
this.admissionLastPrunedAt = now;
|
||||
}
|
||||
}
|
||||
});
|
||||
}, 10_000);
|
||||
this.admissionHeartbeatTimer.unref?.();
|
||||
}
|
||||
|
||||
private schedulePump(immediate = false): void {
|
||||
if (this.inFlight.size > 0 || this.queue.length === 0) return;
|
||||
if (immediate) {
|
||||
@@ -298,10 +674,14 @@ export class RequestLogRuntime {
|
||||
}
|
||||
const commands: QueuedCommand[] = [];
|
||||
let bytes = 0;
|
||||
const isolatedBatch = Boolean(this.queue[0]?.isolated);
|
||||
while (this.queue.length > 0 && commands.length < this.options.batchMaxItems) {
|
||||
const next = this.queue[0];
|
||||
if (commands.length > 0 && bytes + next.batchBytes > this.options.batchMaxBytes) break;
|
||||
commands.push(this.queue.shift()!);
|
||||
if (commands.length > 0 &&
|
||||
(isolatedBatch || next.isolated || bytes + next.batchBytes > this.options.batchMaxBytes)) break;
|
||||
const command = this.queue.shift()!;
|
||||
command.writeAttempts += 1;
|
||||
commands.push(command);
|
||||
bytes += next.batchBytes;
|
||||
}
|
||||
if (commands.length === 0 || !this.writerWorker) return;
|
||||
@@ -351,17 +731,60 @@ export class RequestLogRuntime {
|
||||
this.inFlight.delete(message.batchId);
|
||||
this.outstandingBytes = Math.max(0, this.outstandingBytes - batch.bytes);
|
||||
this.committed += batch.commands.length;
|
||||
this.revision += batch.commands.length;
|
||||
this.markRecordAdmissionsCommitted(batch.commands);
|
||||
this.scheduleRawTraceCleanup(batch.commands);
|
||||
this.schedulePump(true);
|
||||
return;
|
||||
}
|
||||
if (message.type === "batch-error") {
|
||||
this.handleWriterFailure(this.writerWorker, new Error(message.error || "request log batch failed"));
|
||||
this.handleBatchError(message);
|
||||
return;
|
||||
}
|
||||
if (message.type === "maintenance" && (message.updated ?? 0) > 0) {
|
||||
this.revision += 1;
|
||||
return;
|
||||
}
|
||||
settleRpc(this.writerRequests, message);
|
||||
}
|
||||
|
||||
private handleBatchError(message: WorkerResponse): void {
|
||||
if (message.batchId === undefined) {
|
||||
this.handleWriterFailure(this.writerWorker, new Error(message.error || "request log batch failed"));
|
||||
return;
|
||||
}
|
||||
const batch = this.inFlight.get(message.batchId);
|
||||
if (!batch) return;
|
||||
this.inFlight.delete(message.batchId);
|
||||
if (batch.commands.length > 1) {
|
||||
this.queue.unshift(...batch.commands.map((command) => ({ ...command, isolated: true })));
|
||||
this.schedulePump(true);
|
||||
return;
|
||||
}
|
||||
|
||||
const command = batch.commands[0];
|
||||
if (command.writeAttempts < maxCommandWriteAttempts) {
|
||||
this.queue.unshift({ ...command, isolated: true });
|
||||
} else {
|
||||
this.outstandingBytes = Math.max(0, this.outstandingBytes - command.sizeBytes);
|
||||
this.dropped += 1;
|
||||
if (command.kind === "record") {
|
||||
this.rememberRecordAdmission(command.input.requestId, {
|
||||
accepted: false,
|
||||
degraded: false,
|
||||
reason: "writer_unavailable"
|
||||
}, 0, resolveRecordBodyCapturePolicy(command.input));
|
||||
this.scheduleRawTraceCleanup([command]);
|
||||
}
|
||||
console.warn(
|
||||
`[request-log] ${command.kind === "raw-trace-update" ? "Retaining" : "Dropping"} ` +
|
||||
`${command.kind} sequence ${command.sequence} after ` +
|
||||
`${command.writeAttempts} failed write attempts: ${message.error || "request log batch failed"}`
|
||||
);
|
||||
}
|
||||
this.schedulePump(true);
|
||||
}
|
||||
|
||||
private handleWriterFailure(worker: Worker | undefined, error: Error): void {
|
||||
if (!worker || worker !== this.writerWorker) return;
|
||||
this.writerWorker = undefined;
|
||||
@@ -453,7 +876,7 @@ export class RequestLogRuntime {
|
||||
await this.ensureQueryWorker();
|
||||
if (!this.queryWorker) throw new Error("Request log query worker is unavailable.");
|
||||
return await rpc<T>(this.queryWorker, this.queryRequests, ++this.nextRequestId, method, args, {
|
||||
revision: this.committed
|
||||
revision: this.revision
|
||||
});
|
||||
}
|
||||
|
||||
@@ -484,11 +907,20 @@ export function createRequestLogRuntime(options: RequestLogRuntimeOptions): Requ
|
||||
return new RequestLogRuntime(options);
|
||||
}
|
||||
|
||||
function prepareRecordForQueue(input: RequestLogRecordInput, pressure: number): { degraded: boolean; input: RequestLogRecordInput } {
|
||||
function prepareRecordForQueue(
|
||||
input: RequestLogRecordInput,
|
||||
pressure: number
|
||||
): { bodyCaptureMaxBytes: number; degraded: boolean; input: RequestLogRecordInput } {
|
||||
const maxBodyBytes = Math.max(0, Math.min(maxRequestLogBodyBytes, input.maxBodyBytes ?? defaultRequestLogBodyBytes));
|
||||
const ordinarySuccess = input.statusCode >= 200 && input.statusCode < 400 && !input.error;
|
||||
const removeBodies = input.captureBody === false || (pressure >= 0.7 && ordinarySuccess);
|
||||
const bodyCapturePolicy = resolveRecordBodyCapturePolicy(input);
|
||||
const admissionBodyCaptureMaxBytes = bodyCapturePolicy === "none" ||
|
||||
(pressure >= 0.7 && ordinarySuccess)
|
||||
? 0
|
||||
: maxBodyBytes;
|
||||
const removeTrace = pressure >= 0.85 && ordinarySuccess;
|
||||
const suppressTraceBodyValues = (removeBodies || maxBodyBytes === 0) && input.routeTrace !== undefined;
|
||||
const compactedRequest = removeBodies
|
||||
? { buffer: Buffer.alloc(0), compacted: false }
|
||||
: compactBase64ImagePayloads(input.requestBody);
|
||||
@@ -502,17 +934,25 @@ function prepareRecordForQueue(input: RequestLogRecordInput, pressure: number):
|
||||
const responseBodySizeBytes = input.responseBodySizeBytes ?? Buffer.byteLength(input.responseBodyText ?? "");
|
||||
const responseBodyCapturedBytes = Buffer.byteLength(responseBodyText);
|
||||
return {
|
||||
degraded: removeBodies || removeTrace || compactedRequest.compacted || compactedResponse.compacted ||
|
||||
bodyCaptureMaxBytes: admissionBodyCaptureMaxBytes,
|
||||
degraded: removeBodies || removeTrace || suppressTraceBodyValues || compactedRequest.compacted || compactedResponse.compacted ||
|
||||
requestBody.byteLength < compactedRequest.buffer.byteLength ||
|
||||
responseBodyCapturedBytes < responseBodySizeBytes,
|
||||
input: {
|
||||
...input,
|
||||
bodyCapturePolicy,
|
||||
captureBody: !removeBodies,
|
||||
maxBodyBytes: admissionBodyCaptureMaxBytes,
|
||||
requestBody,
|
||||
requestBodySizeBytes: input.requestBodySizeBytes ?? input.requestBody.byteLength,
|
||||
requestBodyTruncated: removeBodies || compactedRequest.compacted || Boolean(input.requestBodyTruncated) ||
|
||||
requestBody.byteLength < compactedRequest.buffer.byteLength,
|
||||
requestHeaders: plainHeaderRecord(input.requestHeaders),
|
||||
routeTrace: removeTrace ? undefined : input.routeTrace,
|
||||
routeTrace: removeTrace
|
||||
? undefined
|
||||
: (suppressTraceBodyValues && input.routeTrace
|
||||
? suppressRouteTraceBodyValues(input.routeTrace)
|
||||
: input.routeTrace),
|
||||
responseBodyText,
|
||||
responseBodySizeBytes,
|
||||
responseBodyTruncated: removeBodies || compactedResponse.compacted || Boolean(input.responseBodyTruncated) ||
|
||||
@@ -522,6 +962,15 @@ function prepareRecordForQueue(input: RequestLogRecordInput, pressure: number):
|
||||
};
|
||||
}
|
||||
|
||||
function resolveRecordBodyCapturePolicy(
|
||||
input: RequestLogRecordInput
|
||||
): "all" | "errors" | "none" {
|
||||
if (input.bodyCapturePolicy === "errors" || input.bodyCapturePolicy === "none") {
|
||||
return input.bodyCapturePolicy;
|
||||
}
|
||||
return input.bodyCapturePolicy === "all" || input.captureBody !== false ? "all" : "none";
|
||||
}
|
||||
|
||||
function prepareRawTraceForQueue(
|
||||
input: RequestLogRawTraceUpdateInput,
|
||||
maxBodyBytes: number
|
||||
@@ -557,6 +1006,46 @@ function prepareRawTraceForQueue(
|
||||
};
|
||||
}
|
||||
|
||||
function constrainRawTraceFiles(
|
||||
files: RequestLogRawTraceFiles | undefined,
|
||||
maxBodyBytes: number
|
||||
): RequestLogRawTraceFiles | undefined {
|
||||
if (!files) return undefined;
|
||||
const {
|
||||
requestBody: _requestBody,
|
||||
responseBody: _responseBody,
|
||||
...metadata
|
||||
} = files;
|
||||
return {
|
||||
...metadata,
|
||||
maxBodyBytes,
|
||||
...(maxBodyBytes > 0 && files.requestBody ? { requestBody: files.requestBody } : {}),
|
||||
...(maxBodyBytes > 0 && files.responseBody ? { responseBody: files.responseBody } : {})
|
||||
};
|
||||
}
|
||||
|
||||
export function suppressRequestLogRawTraceBodies(
|
||||
input: RequestLogRawTraceUpdateInput
|
||||
): RequestLogRawTraceUpdateInput {
|
||||
const requestSize = input.requestBodySizeBytes ??
|
||||
(input.requestBodyText === undefined ? undefined : Buffer.byteLength(input.requestBodyText));
|
||||
const responseSize = input.responseBodySizeBytes ??
|
||||
(input.responseBodyText === undefined ? undefined : Buffer.byteLength(input.responseBodyText));
|
||||
return {
|
||||
...input,
|
||||
...(requestSize === undefined ? {} : {
|
||||
requestBodySizeBytes: requestSize,
|
||||
requestBodyText: "",
|
||||
requestBodyTruncated: Boolean(input.requestBodyTruncated) || requestSize > 0
|
||||
}),
|
||||
...(responseSize === undefined ? {} : {
|
||||
responseBodySizeBytes: responseSize,
|
||||
responseBodyText: "",
|
||||
responseBodyTruncated: Boolean(input.responseBodyTruncated) || responseSize > 0
|
||||
})
|
||||
};
|
||||
}
|
||||
|
||||
function boundedBuffer(value: Buffer, maxBytes: number): Buffer {
|
||||
return value.byteLength <= maxBytes ? value : Buffer.from(value.subarray(0, maxBytes));
|
||||
}
|
||||
@@ -577,7 +1066,7 @@ function plainHeaderRecord(value: Headers | Record<string, string | string[] | u
|
||||
: Object.entries(value).filter((entry): entry is [string, string | string[]] => entry[1] !== undefined);
|
||||
return Object.fromEntries(entries.map(([key, headerValue]) => [
|
||||
key,
|
||||
sensitiveQueueHeaderNames.has(key.toLowerCase()) ? "[redacted]" : headerValue
|
||||
isSensitiveRequestLogHeaderName(key) ? "[redacted]" : headerValue
|
||||
]));
|
||||
}
|
||||
|
||||
@@ -609,7 +1098,13 @@ function jsonBytes(value: unknown): number {
|
||||
}
|
||||
|
||||
function withoutSize(command: QueuedCommand): RequestLogStoreWriteCommand {
|
||||
const { batchBytes: _batchBytes, sizeBytes: _sizeBytes, ...output } = command;
|
||||
const {
|
||||
batchBytes: _batchBytes,
|
||||
isolated: _isolated,
|
||||
sizeBytes: _sizeBytes,
|
||||
writeAttempts: _writeAttempts,
|
||||
...output
|
||||
} = command;
|
||||
return output;
|
||||
}
|
||||
|
||||
@@ -617,6 +1112,10 @@ function positiveInteger(value: number | undefined, fallback: number): number {
|
||||
return Number.isFinite(value) && Number(value) > 0 ? Math.floor(Number(value)) : fallback;
|
||||
}
|
||||
|
||||
function nonNegativeInteger(value: number | undefined): number {
|
||||
return Number.isFinite(value) ? Math.max(0, Math.floor(Number(value))) : 0;
|
||||
}
|
||||
|
||||
function rpc<T>(
|
||||
worker: Worker,
|
||||
requests: Map<number, PendingRpc>,
|
||||
@@ -651,6 +1150,12 @@ function errorCode(error: unknown): string | undefined {
|
||||
: undefined;
|
||||
}
|
||||
|
||||
function isTransientSqliteLock(error: unknown): boolean {
|
||||
const code = errorCode(error);
|
||||
return code === "SQLITE_BUSY" || code === "SQLITE_LOCKED" ||
|
||||
(error instanceof Error && /database is (?:busy|locked)/i.test(error.message));
|
||||
}
|
||||
|
||||
function formatRuntimeError(error: unknown): string {
|
||||
return error instanceof Error ? error.stack || error.message : String(error);
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -11,6 +11,7 @@ import {
|
||||
} from "@ccr/core/observability/request-log-store";
|
||||
import { resolveRawTraceBodyLimit } from "@ccr/core/observability/request-log-limits";
|
||||
import { compactBase64ImagePayloads } from "@ccr/core/observability/request-log-body";
|
||||
import { preloadUsagePriceCatalog } from "@ccr/core/models/pricing-service";
|
||||
|
||||
type WorkerConfiguration = {
|
||||
dbFile: string;
|
||||
@@ -31,7 +32,10 @@ type WorkerMessage = {
|
||||
const configuration = workerData as WorkerConfiguration;
|
||||
const store = new RequestLogStore(configuration.dbFile);
|
||||
let chain = Promise.resolve();
|
||||
let pricingBackfillActive = false;
|
||||
let pricingRefreshPromise: Promise<void> | undefined;
|
||||
let queryRevision = -1;
|
||||
let shuttingDown = false;
|
||||
|
||||
void store.initialize().then(() => {
|
||||
parentPort?.postMessage({ type: "ready" });
|
||||
@@ -53,8 +57,9 @@ async function handleMessage(message: WorkerMessage): Promise<void> {
|
||||
if (message.type === "batch") {
|
||||
if (configuration.mode !== "writer") throw new Error("Query worker cannot process writes.");
|
||||
const commands = message.commands ?? [];
|
||||
await store.writeBatch(commands.map(reviveCommand));
|
||||
const result = await store.writeBatch(commands.map(reviveCommand));
|
||||
parentPort?.postMessage({ batchId: message.batchId, type: "ack" });
|
||||
if (result.pricingRefreshNeeded) schedulePricingRefresh();
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -82,6 +87,7 @@ async function handleMessage(message: WorkerMessage): Promise<void> {
|
||||
result = await store.list(args[0] as Parameters<RequestLogStore["list"]>[0]);
|
||||
break;
|
||||
case "shutdown":
|
||||
shuttingDown = true;
|
||||
await store.close();
|
||||
result = true;
|
||||
break;
|
||||
@@ -92,6 +98,44 @@ async function handleMessage(message: WorkerMessage): Promise<void> {
|
||||
if (message.method === "shutdown") parentPort?.close();
|
||||
}
|
||||
|
||||
function schedulePricingRefresh(): void {
|
||||
if (shuttingDown || pricingRefreshPromise) return;
|
||||
pricingRefreshPromise = preloadUsagePriceCatalog()
|
||||
.then(() => {
|
||||
if (shuttingDown) return;
|
||||
schedulePricingBackfillPage();
|
||||
})
|
||||
.catch((error) => {
|
||||
console.warn(`[request-log] Failed to refresh pricing catalog: ${formatError(error)}`);
|
||||
})
|
||||
.finally(() => {
|
||||
pricingRefreshPromise = undefined;
|
||||
});
|
||||
}
|
||||
|
||||
function schedulePricingBackfillPage(beforeId?: number): void {
|
||||
if (shuttingDown || (pricingBackfillActive && beforeId === undefined)) return;
|
||||
pricingBackfillActive = true;
|
||||
chain = chain.then(async () => {
|
||||
if (shuttingDown) {
|
||||
pricingBackfillActive = false;
|
||||
return;
|
||||
}
|
||||
const page = await store.backfillMissingUsageCostsPage({ beforeId });
|
||||
if (page.updated > 0) parentPort?.postMessage({ type: "maintenance", updated: page.updated });
|
||||
if (page.nextBeforeId === undefined) {
|
||||
pricingBackfillActive = false;
|
||||
return;
|
||||
}
|
||||
// Yield between pages so normal log writes already queued by the parent can
|
||||
// run before the next maintenance batch.
|
||||
setImmediate(() => schedulePricingBackfillPage(page.nextBeforeId));
|
||||
}).catch((error) => {
|
||||
pricingBackfillActive = false;
|
||||
console.warn(`[request-log] Failed to backfill usage costs: ${formatError(error)}`);
|
||||
});
|
||||
}
|
||||
|
||||
function reviveCommand(command: RequestLogStoreWriteCommand): RequestLogStoreWriteCommand {
|
||||
if (command.kind === "raw-trace-update") {
|
||||
const input = { ...command.input };
|
||||
|
||||
@@ -7,6 +7,7 @@ import type {
|
||||
RequestRouteTracePhase,
|
||||
RequestRouteTraceTarget
|
||||
} from "@ccr/core/contracts/app";
|
||||
import { isSensitiveRequestLogHeaderName } from "@ccr/core/observability/sensitive-headers";
|
||||
|
||||
export type RouteTraceObservation = {
|
||||
attempt?: number;
|
||||
@@ -115,15 +116,18 @@ export class RequestRouteTraceRecorder implements RouteTraceObserver {
|
||||
this.pushHop(hop);
|
||||
}
|
||||
|
||||
finish(): RequestRouteTrace {
|
||||
finish(options: { captureBodyValues?: boolean } = {}): RequestRouteTrace {
|
||||
if (this.finished) {
|
||||
return this.finished;
|
||||
}
|
||||
const hops = options.captureBodyValues === false
|
||||
? this.hops.map(suppressRouteTraceHopBodyValues)
|
||||
: this.hops;
|
||||
this.finished = {
|
||||
attemptCount: this.attempts.size,
|
||||
complete: true,
|
||||
hopCount: this.hops.length,
|
||||
hops: this.hops,
|
||||
hopCount: hops.length,
|
||||
hops,
|
||||
truncated: this.truncated,
|
||||
version: 2
|
||||
};
|
||||
@@ -172,10 +176,31 @@ export class RequestRouteTraceRecorder implements RouteTraceObserver {
|
||||
}
|
||||
}
|
||||
|
||||
export function suppressRouteTraceBodyValues(trace: RequestRouteTrace): RequestRouteTrace {
|
||||
const hops = trace.hops.map(suppressRouteTraceHopBodyValues);
|
||||
return hops.every((hop, index) => hop === trace.hops[index])
|
||||
? trace
|
||||
: { ...trace, hops };
|
||||
}
|
||||
|
||||
function boundedObservationValue<T extends object>(value: T): T {
|
||||
return previewValue(value).value as T;
|
||||
}
|
||||
|
||||
function suppressRouteTraceHopBodyValues(hop: RequestRouteTraceHop): RequestRouteTraceHop {
|
||||
if (!hop.changes.some((change) => change.scope === "body")) {
|
||||
return hop;
|
||||
}
|
||||
return {
|
||||
...hop,
|
||||
changes: hop.changes.map((change) => {
|
||||
if (change.scope !== "body") return change;
|
||||
const { after: _after, before: _before, ...metadata } = change;
|
||||
return metadata;
|
||||
})
|
||||
};
|
||||
}
|
||||
|
||||
function sanitizeReportedChange(change: RequestRouteTraceChange): RequestRouteTraceChange {
|
||||
const path = normalizePath(change.path);
|
||||
const redacted = Boolean(change.redacted) || pathContainsSensitiveName(path);
|
||||
@@ -207,7 +232,7 @@ function sanitizeUrlValue(value: unknown): unknown {
|
||||
try {
|
||||
const url = new URL(value, "http://127.0.0.1");
|
||||
for (const key of [...url.searchParams.keys()]) {
|
||||
if (sensitiveNames.test(key) || /^(?:key|token)$/i.test(key)) {
|
||||
if (isSensitiveName(key) || /^(?:key|token)$/i.test(key)) {
|
||||
url.searchParams.set(key, redactedDisplayValue);
|
||||
}
|
||||
}
|
||||
@@ -239,7 +264,7 @@ function boundedPreview(
|
||||
budget.truncated = true;
|
||||
return truncatedDisplayValue;
|
||||
}
|
||||
if (key && sensitiveNames.test(key)) {
|
||||
if (key && isSensitiveName(key)) {
|
||||
budget.remaining -= redactedDisplayValue.length;
|
||||
return redactedDisplayValue;
|
||||
}
|
||||
@@ -323,7 +348,11 @@ function pathContainsSensitiveName(path: string): boolean {
|
||||
.split("/")
|
||||
.filter(Boolean)
|
||||
.map((part) => part.replaceAll("~1", "/").replaceAll("~0", "~"))
|
||||
.some((part) => sensitiveNames.test(part));
|
||||
.some(isSensitiveName);
|
||||
}
|
||||
|
||||
function isSensitiveName(value: string): boolean {
|
||||
return sensitiveNames.test(value) || isSensitiveRequestLogHeaderName(value);
|
||||
}
|
||||
|
||||
function normalizePath(value: string): string {
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
export const sensitiveRequestLogHeaderNames: ReadonlySet<string> = new Set([
|
||||
"api-key",
|
||||
"authorization",
|
||||
"cookie",
|
||||
"ocp-apim-subscription-key",
|
||||
"proxy-authorization",
|
||||
"set-cookie",
|
||||
"x-api-key",
|
||||
"x-auth-api-key-id",
|
||||
"x-auth-sub",
|
||||
"x-goog-api-key"
|
||||
]);
|
||||
|
||||
// Provider plugins may use arbitrary authentication header names. Match
|
||||
// security-bearing name segments in addition to the compatibility allowlist
|
||||
// so a newly introduced x-*-token/secret/key header is fail-closed by default.
|
||||
const sensitiveRequestLogHeaderPattern =
|
||||
/(?:^|[-_.])(?:auth(?:orization)?|bearer|cookie|credential|csrf|jwt|key|pass(?:word|wd)?|secret|signature|token)(?:$|[-_.])/i;
|
||||
|
||||
export function isSensitiveRequestLogHeaderName(value: string): boolean {
|
||||
const normalized = value.trim().toLowerCase();
|
||||
return sensitiveRequestLogHeaderNames.has(normalized) || sensitiveRequestLogHeaderPattern.test(normalized);
|
||||
}
|
||||
@@ -6,15 +6,24 @@ export type {
|
||||
Statement as BetterSqliteStatement
|
||||
} from "better-sqlite3";
|
||||
|
||||
export type BetterSqliteDatabaseOptions = {
|
||||
fileMustExist?: boolean;
|
||||
readonly?: boolean;
|
||||
timeout?: number;
|
||||
};
|
||||
|
||||
const requireFromHere = createRequire(__filename);
|
||||
let resolvedNativeBinding: string | undefined;
|
||||
let nativeBindingResolved = false;
|
||||
|
||||
export function createBetterSqliteDatabase(filename: string): BetterSqliteDatabase {
|
||||
export function createBetterSqliteDatabase(
|
||||
filename: string,
|
||||
options: BetterSqliteDatabaseOptions = {}
|
||||
): BetterSqliteDatabase {
|
||||
const nativeBinding = resolveBetterSqliteNativeBinding();
|
||||
return nativeBinding
|
||||
? new DatabaseConstructor(filename, { nativeBinding })
|
||||
: new DatabaseConstructor(filename);
|
||||
? new DatabaseConstructor(filename, { ...options, nativeBinding })
|
||||
: new DatabaseConstructor(filename, options);
|
||||
}
|
||||
|
||||
function resolveBetterSqliteNativeBinding(): string | undefined {
|
||||
|
||||
+136
@@ -0,0 +1,136 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import test from "node:test";
|
||||
import { preloadUsagePriceCatalog } from "@ccr/core/models/pricing-service.ts";
|
||||
import { RequestLogStore } from "@ccr/core/observability/request-log-store.ts";
|
||||
import { createBetterSqliteDatabase } from "@ccr/core/storage/sqlite-native.ts";
|
||||
|
||||
test("RequestLogStore commits while a background pricing refresh is stalled", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-pricing-lock-test-"));
|
||||
const dbFile = path.join(dir, "request-logs.sqlite");
|
||||
const first = new RequestLogStore(dbFile);
|
||||
const second = new RequestLogStore(dbFile);
|
||||
const previousFetch = globalThis.fetch;
|
||||
let releaseFetch;
|
||||
let signalFetchStarted;
|
||||
const fetchGate = new Promise((resolve) => {
|
||||
releaseFetch = resolve;
|
||||
});
|
||||
const fetchStarted = new Promise((resolve) => {
|
||||
signalFetchStarted = resolve;
|
||||
});
|
||||
globalThis.fetch = async () => {
|
||||
signalFetchStarted();
|
||||
await fetchGate;
|
||||
return new Response(JSON.stringify({
|
||||
"pricing-lock-model": {
|
||||
input_cost_per_token: 0.000001,
|
||||
litellm_provider: "pricing-lock-provider",
|
||||
output_cost_per_token: 0.000002
|
||||
}
|
||||
}), {
|
||||
headers: { "content-type": "application/json" },
|
||||
status: 200
|
||||
});
|
||||
};
|
||||
|
||||
let pricingRefresh;
|
||||
let fetchStartTimeout;
|
||||
try {
|
||||
pricingRefresh = preloadUsagePriceCatalog();
|
||||
await Promise.race([
|
||||
fetchStarted,
|
||||
new Promise((_, reject) => {
|
||||
fetchStartTimeout = setTimeout(() => reject(new Error("pricing fetch did not start")), 5_000);
|
||||
})
|
||||
]);
|
||||
clearTimeout(fetchStartTimeout);
|
||||
fetchStartTimeout = undefined;
|
||||
|
||||
const pricedWrite = first.writeBatch([{
|
||||
eventId: "priced-event",
|
||||
input: createRecord("priced-request", '{"usage":{"input_tokens":1,"output_tokens":1}}'),
|
||||
kind: "record",
|
||||
sequence: 1
|
||||
}]);
|
||||
const pricedResult = await Promise.race([
|
||||
pricedWrite,
|
||||
new Promise((_, reject) => {
|
||||
fetchStartTimeout = setTimeout(() => reject(new Error("request log write waited for pricing")), 1_000);
|
||||
})
|
||||
]);
|
||||
assert.equal(pricedResult.pricingRefreshNeeded, true);
|
||||
clearTimeout(fetchStartTimeout);
|
||||
fetchStartTimeout = undefined;
|
||||
|
||||
// The pricing request is deliberately stalled. Both writers can commit,
|
||||
// and cost can be filled after the shared catalog refresh completes.
|
||||
const zeroTokenResult = await second.writeBatch([{
|
||||
eventId: "zero-token-event",
|
||||
input: createRecord("zero-token-request", "{}"),
|
||||
kind: "record",
|
||||
sequence: 2
|
||||
}]);
|
||||
assert.equal(zeroTokenResult.pricingRefreshNeeded, false);
|
||||
|
||||
const page = await first.list({ pageSize: 25 });
|
||||
assert.equal(page.items.length, 2);
|
||||
releaseFetch();
|
||||
await pricingRefresh;
|
||||
// Fill the newest page with models that cannot be priced. The older known
|
||||
// model must still be reached by pagination.
|
||||
const database = createBetterSqliteDatabase(dbFile);
|
||||
try {
|
||||
const insert = database.prepare(`
|
||||
INSERT INTO request_logs (
|
||||
created_at,
|
||||
method,
|
||||
path,
|
||||
model,
|
||||
provider,
|
||||
input_tokens,
|
||||
total_tokens
|
||||
) VALUES (?, 'POST', '/v1/messages', 'unknown-price-model', 'unknown-provider', 1, 1)
|
||||
`);
|
||||
database.transaction(() => {
|
||||
for (let index = 0; index < 1_001; index += 1) {
|
||||
insert.run(new Date().toISOString());
|
||||
}
|
||||
})();
|
||||
} finally {
|
||||
database.close();
|
||||
}
|
||||
assert.equal(await first.backfillMissingUsageCosts(), 1);
|
||||
const updated = await first.getDetail({ id: page.items.find((item) => item.requestId === "priced-request").id });
|
||||
assert.equal(updated.costUsd, 0.000003);
|
||||
} finally {
|
||||
clearTimeout(fetchStartTimeout);
|
||||
releaseFetch?.();
|
||||
await pricingRefresh?.catch(() => undefined);
|
||||
globalThis.fetch = previousFetch;
|
||||
await Promise.all([first.close(), second.close()]);
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
function createRecord(requestId, responseBodyText) {
|
||||
const now = new Date().toISOString();
|
||||
return {
|
||||
completedAt: now,
|
||||
durationMs: 10,
|
||||
method: "POST",
|
||||
model: "pricing-lock-model",
|
||||
path: "/v1/messages",
|
||||
providerName: "pricing-lock-provider",
|
||||
requestBody: Buffer.from('{"model":"pricing-lock-model"}'),
|
||||
requestHeaders: { "content-type": "application/json" },
|
||||
requestId,
|
||||
responseBodyText,
|
||||
responseHeaders: { "content-type": "application/json" },
|
||||
startedAt: now,
|
||||
statusCode: 200,
|
||||
url: "http://127.0.0.1:3456/v1/messages"
|
||||
};
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -13,6 +13,104 @@ import { createBetterSqliteDatabase } from "@ccr/core/storage/sqlite-native.ts";
|
||||
const execFileAsync = promisify(execFile);
|
||||
const isBoundedHeapWorker = process.env.CCR_REQUEST_LOG_BOUNDED_HEAP_WORKER === "1";
|
||||
|
||||
test("RequestLogStore resumes interrupted gateway migrations across bounded batches", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-paged-migration-test-"));
|
||||
const dbFile = path.join(dir, "request-logs.sqlite");
|
||||
let store;
|
||||
try {
|
||||
const legacy = createBetterSqliteDatabase(dbFile);
|
||||
try {
|
||||
legacy.exec(`
|
||||
CREATE TABLE request_logs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
created_at TEXT NOT NULL,
|
||||
event_id TEXT NOT NULL DEFAULT '',
|
||||
status_code INTEGER NOT NULL DEFAULT 0,
|
||||
ok INTEGER NOT NULL DEFAULT 0,
|
||||
error TEXT NOT NULL DEFAULT '',
|
||||
gateway_status_code INTEGER NOT NULL DEFAULT 0,
|
||||
gateway_ok INTEGER NOT NULL DEFAULT 0,
|
||||
gateway_error TEXT NOT NULL DEFAULT '',
|
||||
gateway_final_attempt INTEGER NOT NULL DEFAULT 1,
|
||||
response_headers TEXT NOT NULL DEFAULT '{}'
|
||||
);
|
||||
CREATE TABLE request_log_schema_migrations (
|
||||
migration TEXT PRIMARY KEY,
|
||||
last_id INTEGER NOT NULL DEFAULT 0,
|
||||
completed INTEGER NOT NULL DEFAULT 0,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
`);
|
||||
const insert = legacy.prepare(`
|
||||
INSERT INTO request_logs (created_at, event_id, status_code, ok, error, response_headers)
|
||||
VALUES (?, ?, 503, 0, 'legacy failure', ?)
|
||||
`);
|
||||
const createdAt = new Date().toISOString();
|
||||
legacy.transaction(() => {
|
||||
for (let index = 0; index < 1_205; index += 1) {
|
||||
insert.run(
|
||||
createdAt,
|
||||
`legacy-event-${index}`,
|
||||
JSON.stringify({ "x-ccr-fallback-attempts": String((index % 4) + 1) })
|
||||
);
|
||||
}
|
||||
})();
|
||||
legacy.exec(`
|
||||
UPDATE request_logs
|
||||
SET gateway_final_attempt = ((id - 1) % 4) + 1
|
||||
WHERE id <= 500;
|
||||
INSERT INTO request_log_schema_migrations (migration, last_id, completed, updated_at)
|
||||
VALUES ('gateway-final-attempt-v1', 500, 0, 1);
|
||||
`);
|
||||
} finally {
|
||||
legacy.close();
|
||||
}
|
||||
|
||||
store = new RequestLogStore(dbFile);
|
||||
await store.list({ pageSize: 1 });
|
||||
|
||||
const migrated = createBetterSqliteDatabase(dbFile);
|
||||
try {
|
||||
const rows = migrated.prepare(`
|
||||
SELECT gateway_final_attempt AS attempt, COUNT(*) AS total
|
||||
FROM request_logs
|
||||
GROUP BY gateway_final_attempt
|
||||
ORDER BY gateway_final_attempt
|
||||
`).all();
|
||||
assert.deepEqual(rows.map((row) => [row.attempt, row.total]), [
|
||||
[1, 302],
|
||||
[2, 301],
|
||||
[3, 301],
|
||||
[4, 301]
|
||||
]);
|
||||
const outcome = migrated.prepare(`
|
||||
SELECT COUNT(*) AS total
|
||||
FROM request_logs
|
||||
WHERE gateway_status_code = 503
|
||||
AND gateway_ok = 0
|
||||
AND gateway_error = 'legacy failure'
|
||||
`).get();
|
||||
assert.equal(outcome.total, 1_205);
|
||||
const migrations = migrated.prepare(`
|
||||
SELECT migration, completed
|
||||
FROM request_log_schema_migrations
|
||||
ORDER BY migration
|
||||
`).all();
|
||||
assert.deepEqual(migrations, [
|
||||
{ completed: 1, migration: "gateway-final-attempt-v1" },
|
||||
{ completed: 1, migration: "gateway-outcome-v1" }
|
||||
]);
|
||||
const indexes = migrated.prepare("PRAGMA index_list(request_logs)").all();
|
||||
assert.equal(indexes.some((index) => index.name === "request_logs_request_id_idx"), true);
|
||||
} finally {
|
||||
migrated.close();
|
||||
}
|
||||
} finally {
|
||||
await store?.close();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
async function recordLargeAgentRequests(store, dbFile, { paddingBytes, requestCount, sessionId }) {
|
||||
const padding = "x".repeat(paddingBytes);
|
||||
const startedAt = new Date().toISOString();
|
||||
@@ -232,17 +330,23 @@ test("RequestLogStore redacts secrets and records CCR metadata", async () => {
|
||||
requestBody: Buffer.from(JSON.stringify({ model: "gpt-test", stream: true }), "utf8"),
|
||||
requestHeaders: {
|
||||
accept: "text/event-stream",
|
||||
"api-key": "azure-request-secret",
|
||||
authorization: "Bearer request-secret",
|
||||
cookie: "session=request-secret",
|
||||
"content-type": "application/json",
|
||||
"ocp-apim-subscription-key": "bing-request-secret",
|
||||
"x-amz-security-token": "aws-request-secret",
|
||||
"x-auth-token": "custom-request-secret",
|
||||
"x-ccr-provider-credential-chain": "cred-a, cred-b",
|
||||
"x-ccr-provider-credential-id": "cred-a"
|
||||
"x-ccr-provider-credential-id": "cred-a",
|
||||
"x-goog-api-key": "google-request-secret"
|
||||
},
|
||||
requestId: "request-log-metadata-test",
|
||||
responseBodyText: "",
|
||||
responseHeaders: {
|
||||
"content-type": "text/event-stream",
|
||||
"x-api-key": "response-secret",
|
||||
"x-company-client-secret": "custom-response-secret",
|
||||
"x-ccr-provider-credential-saturated": "true",
|
||||
"x-gateway-billing-cache-read-tokens": "10",
|
||||
"x-gateway-billing-input-tokens": "100",
|
||||
@@ -258,9 +362,15 @@ test("RequestLogStore redacts secrets and records CCR metadata", async () => {
|
||||
const detail = await store.getDetail({ id: page.items[0].id });
|
||||
|
||||
assert.ok(detail);
|
||||
assert.equal(detail.requestHeaders["api-key"], "[redacted]");
|
||||
assert.equal(detail.requestHeaders.authorization, "[redacted]");
|
||||
assert.equal(detail.requestHeaders.cookie, "[redacted]");
|
||||
assert.equal(detail.requestHeaders["ocp-apim-subscription-key"], "[redacted]");
|
||||
assert.equal(detail.requestHeaders["x-amz-security-token"], "[redacted]");
|
||||
assert.equal(detail.requestHeaders["x-auth-token"], "[redacted]");
|
||||
assert.equal(detail.requestHeaders["x-goog-api-key"], "[redacted]");
|
||||
assert.equal(detail.responseHeaders["x-api-key"], "[redacted]");
|
||||
assert.equal(detail.responseHeaders["x-company-client-secret"], "[redacted]");
|
||||
assert.equal(detail.credentialId, "cred-a");
|
||||
assert.deepEqual(detail.credentialChain, ["cred-a", "cred-b"]);
|
||||
assert.equal(detail.credentialSaturated, true);
|
||||
@@ -313,6 +423,257 @@ test("RequestLogStore marks interrupted successful-status streams as errors", as
|
||||
}
|
||||
});
|
||||
|
||||
test("RequestLogStore keeps an authoritative gateway failure when a later raw trace reports HTTP 200", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-raw-authority-test-"));
|
||||
const store = new RequestLogStore(path.join(dir, "request-logs.sqlite"));
|
||||
const startedAt = new Date().toISOString();
|
||||
try {
|
||||
await store.record({
|
||||
completedAt: startedAt,
|
||||
durationMs: 25,
|
||||
error: "Client connection closed before response completed.",
|
||||
method: "POST",
|
||||
path: "/v1/messages",
|
||||
providerName: "gateway-provider",
|
||||
requestBody: Buffer.from('{"model":"gateway-model","stream":true}'),
|
||||
requestHeaders: { "content-type": "application/json" },
|
||||
requestId: "gateway-failure-before-raw",
|
||||
responseBodyText: "gateway-captured-error-body",
|
||||
responseHeaders: { "content-type": "text/event-stream" },
|
||||
startedAt,
|
||||
statusCode: 499,
|
||||
url: "http://127.0.0.1:3456/v1/messages"
|
||||
});
|
||||
|
||||
await store.updateFromRawTrace({
|
||||
bodyCapturePolicy: "errors",
|
||||
isStream: true,
|
||||
requestId: "gateway-failure-before-raw",
|
||||
responseBodySizeBytes: 128,
|
||||
responseBodyText: "",
|
||||
responseBodyTruncated: true,
|
||||
statusCode: 200
|
||||
});
|
||||
|
||||
const page = await store.list({ pageSize: 25 });
|
||||
const detail = await store.getDetail({ id: page.items[0].id });
|
||||
assert.equal(detail.statusCode, 499);
|
||||
assert.equal(detail.ok, false);
|
||||
assert.equal(detail.error, "Client connection closed before response completed.");
|
||||
assert.equal(detail.responseBody.text, "gateway-captured-error-body");
|
||||
} finally {
|
||||
await store.close();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("RequestLogStore keeps an errorless gateway HTTP 500 authoritative over raw HTTP 200", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-status-authority-test-"));
|
||||
const store = new RequestLogStore(path.join(dir, "request-logs.sqlite"));
|
||||
const startedAt = new Date().toISOString();
|
||||
try {
|
||||
await store.record({
|
||||
completedAt: startedAt,
|
||||
durationMs: 25,
|
||||
method: "POST",
|
||||
path: "/v1/messages",
|
||||
providerName: "gateway-provider",
|
||||
requestBody: Buffer.from('{"model":"gateway-model"}'),
|
||||
requestHeaders: { "content-type": "application/json" },
|
||||
requestId: "gateway-status-failure-before-raw",
|
||||
responseBodyText: '{"type":"gateway_failure"}',
|
||||
responseHeaders: { "content-type": "application/json" },
|
||||
startedAt,
|
||||
statusCode: 500,
|
||||
url: "http://127.0.0.1:3456/v1/messages"
|
||||
});
|
||||
|
||||
await store.updateFromRawTrace({
|
||||
requestId: "gateway-status-failure-before-raw",
|
||||
responseBodyText: '{"type":"upstream_success"}',
|
||||
statusCode: 200
|
||||
});
|
||||
|
||||
const page = await store.list({ pageSize: 25 });
|
||||
const detail = await store.getDetail({ id: page.items[0].id });
|
||||
assert.equal(detail.statusCode, 500);
|
||||
assert.equal(detail.ok, false);
|
||||
} finally {
|
||||
await store.close();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("RequestLogStore consumes fallback bundles by unique bundle id and only final attempt mutates outcome", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-fallback-bundles-test-"));
|
||||
const dbFile = path.join(dir, "request-logs.sqlite");
|
||||
const store = new RequestLogStore(dbFile);
|
||||
const startedAt = new Date().toISOString();
|
||||
try {
|
||||
for (const requestId of ["fallback-final-success", "fallback-final-failure"]) {
|
||||
await store.record({
|
||||
completedAt: startedAt,
|
||||
durationMs: 25,
|
||||
method: "POST",
|
||||
path: "/v1/messages",
|
||||
providerName: "gateway-provider",
|
||||
requestBody: Buffer.from('{"model":"gateway-model"}'),
|
||||
requestHeaders: { "content-type": "application/json" },
|
||||
requestId,
|
||||
responseBodyText: "gateway-body",
|
||||
responseHeaders: {
|
||||
"content-type": "application/json",
|
||||
"x-ccr-fallback-attempts": "2"
|
||||
},
|
||||
startedAt,
|
||||
statusCode: 200,
|
||||
url: "http://127.0.0.1:3456/v1/messages"
|
||||
});
|
||||
}
|
||||
|
||||
await store.writeBatch([
|
||||
{
|
||||
input: {
|
||||
attempt: 2,
|
||||
bundleId: "success-final-bundle",
|
||||
requestId: "fallback-final-success",
|
||||
responseBodyText: "final-success-body",
|
||||
statusCode: 200
|
||||
},
|
||||
kind: "raw-trace-update",
|
||||
sequence: 1
|
||||
},
|
||||
{
|
||||
input: {
|
||||
attempt: 1,
|
||||
bundleId: "success-intermediate-bundle",
|
||||
requestId: "fallback-final-success",
|
||||
responseBodyText: "intermediate-failure-body",
|
||||
statusCode: 500
|
||||
},
|
||||
kind: "raw-trace-update",
|
||||
sequence: 2
|
||||
},
|
||||
{
|
||||
input: {
|
||||
attempt: 2,
|
||||
bundleId: "failure-final-bundle",
|
||||
requestId: "fallback-final-failure",
|
||||
responseBodyText: "final-failure-body",
|
||||
statusCode: 502
|
||||
},
|
||||
kind: "raw-trace-update",
|
||||
sequence: 3
|
||||
},
|
||||
{
|
||||
input: {
|
||||
attempt: 1,
|
||||
bundleId: "failure-intermediate-bundle",
|
||||
requestId: "fallback-final-failure",
|
||||
responseBodyText: "intermediate-success-body",
|
||||
statusCode: 200
|
||||
},
|
||||
kind: "raw-trace-update",
|
||||
sequence: 4
|
||||
},
|
||||
{
|
||||
input: {
|
||||
attempt: 2,
|
||||
bundleId: "success-final-bundle",
|
||||
requestId: "fallback-final-success",
|
||||
responseBodyText: "duplicate-must-not-apply",
|
||||
statusCode: 503
|
||||
},
|
||||
kind: "raw-trace-update",
|
||||
sequence: 5
|
||||
}
|
||||
]);
|
||||
|
||||
const page = await store.list({ pageSize: 25 });
|
||||
const success = await store.getDetail({
|
||||
id: page.items.find((item) => item.requestId === "fallback-final-success").id
|
||||
});
|
||||
const failure = await store.getDetail({
|
||||
id: page.items.find((item) => item.requestId === "fallback-final-failure").id
|
||||
});
|
||||
assert.equal(success.statusCode, 200);
|
||||
assert.equal(success.ok, true);
|
||||
assert.equal(success.responseBody.text, "final-success-body");
|
||||
assert.equal(failure.statusCode, 502);
|
||||
assert.equal(failure.ok, false);
|
||||
assert.equal(failure.responseBody.text, "final-failure-body");
|
||||
|
||||
const database = createBetterSqliteDatabase(dbFile);
|
||||
try {
|
||||
const count = database.prepare("SELECT COUNT(*) AS total FROM request_log_raw_trace_events").get();
|
||||
assert.equal(Number(count.total), 4);
|
||||
} finally {
|
||||
database.close();
|
||||
}
|
||||
} finally {
|
||||
await store.close();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("RequestLogStore detects raw errors before applying errors-only body suppression", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-raw-error-policy-test-"));
|
||||
const store = new RequestLogStore(path.join(dir, "request-logs.sqlite"));
|
||||
const startedAt = new Date().toISOString();
|
||||
try {
|
||||
for (const requestId of ["raw-http-error", "raw-sse-error"]) {
|
||||
await store.record({
|
||||
completedAt: startedAt,
|
||||
durationMs: 25,
|
||||
method: "POST",
|
||||
path: "/v1/messages",
|
||||
providerName: "gateway-provider",
|
||||
requestBody: Buffer.from('{"model":"gateway-model"}'),
|
||||
requestHeaders: { "content-type": "application/json" },
|
||||
requestId,
|
||||
responseBodyText: "gateway-success-body",
|
||||
responseHeaders: { "content-type": "application/json" },
|
||||
startedAt,
|
||||
statusCode: 200,
|
||||
url: "http://127.0.0.1:3456/v1/messages"
|
||||
});
|
||||
}
|
||||
|
||||
await store.updateFromRawTrace({
|
||||
bodyCapturePolicy: "errors",
|
||||
requestId: "raw-http-error",
|
||||
responseBodyContentType: "application/json",
|
||||
responseBodyText: '{"error":{"message":"upstream failed"}}',
|
||||
responseHeaders: { "content-type": "application/json" },
|
||||
statusCode: 500
|
||||
});
|
||||
await store.updateFromRawTrace({
|
||||
bodyCapturePolicy: "errors",
|
||||
isStream: true,
|
||||
requestId: "raw-sse-error",
|
||||
responseBodyContentType: "text/event-stream",
|
||||
responseBodyText: 'event: error\ndata: {"error":{"message":"late failure"}}\n\n',
|
||||
responseHeaders: { "content-type": "text/event-stream" },
|
||||
statusCode: 200
|
||||
});
|
||||
|
||||
const page = await store.list({ pageSize: 25 });
|
||||
const httpEntry = page.items.find((item) => item.requestId === "raw-http-error");
|
||||
const sseEntry = page.items.find((item) => item.requestId === "raw-sse-error");
|
||||
const httpDetail = await store.getDetail({ id: httpEntry.id });
|
||||
const sseDetail = await store.getDetail({ id: sseEntry.id });
|
||||
assert.equal(httpDetail.ok, false);
|
||||
assert.equal(httpDetail.statusCode, 500);
|
||||
assert.match(httpDetail.responseBody.text, /upstream failed/);
|
||||
assert.equal(sseDetail.ok, false);
|
||||
assert.match(sseDetail.error, /late failure/);
|
||||
assert.match(sseDetail.responseBody.text, /late failure/);
|
||||
} finally {
|
||||
await store.close();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("RequestLogStore applies raw trace updates to existing request logs", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-raw-trace-test-"));
|
||||
try {
|
||||
@@ -344,7 +705,13 @@ test("RequestLogStore applies raw trace updates to existing request logs", async
|
||||
isStream: true,
|
||||
model: "trace-model",
|
||||
provider: "trace-provider",
|
||||
requestHeaders: { "x-client-name": "codex-cli" },
|
||||
requestHeaders: {
|
||||
"api-key": "raw-azure-secret",
|
||||
"ocp-apim-subscription-key": "raw-bing-secret",
|
||||
"x-ccr-provider-credential-id": "raw-credential-id",
|
||||
"x-client-name": "codex-cli",
|
||||
"x-goog-api-key": "raw-google-secret"
|
||||
},
|
||||
requestId: "raw-trace-request",
|
||||
responseBodyContentType: "text/event-stream",
|
||||
responseBodyText: errorStream,
|
||||
@@ -360,11 +727,16 @@ test("RequestLogStore applies raw trace updates to existing request logs", async
|
||||
assert.ok(detail);
|
||||
assert.equal(detail.model, "trace-model");
|
||||
assert.equal(detail.provider, "trace-provider");
|
||||
assert.equal(detail.credentialId, "raw-credential-id");
|
||||
assert.equal(detail.ok, false);
|
||||
assert.equal(detail.isStream, true);
|
||||
assert.match(detail.error, /rate_limit_error: quota exceeded/);
|
||||
assert.match(detail.responseBody?.text ?? "", /quota exceeded/);
|
||||
assert.equal(detail.requestHeaders["api-key"], "[redacted]");
|
||||
assert.equal(detail.requestHeaders["ocp-apim-subscription-key"], "[redacted]");
|
||||
assert.equal(detail.requestHeaders["x-ccr-provider-credential-id"], "[redacted]");
|
||||
assert.equal(detail.requestHeaders["x-client-name"], "codex-cli");
|
||||
assert.equal(detail.requestHeaders["x-goog-api-key"], "[redacted]");
|
||||
} finally {
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import assert from "node:assert/strict";
|
||||
import path from "node:path";
|
||||
import test from "node:test";
|
||||
import { createDefaultAppConfig } from "@ccr/core/config/default-config.ts";
|
||||
import {
|
||||
@@ -6,9 +7,16 @@ import {
|
||||
BROWSER_AUTOMATION_MCP_PATH,
|
||||
BROWSER_AUTOMATION_MCP_SERVER_NAME,
|
||||
browserAutomationMcpEnabled,
|
||||
bundledToolHubMcpEntryPathCandidates,
|
||||
toolHubMcpRuntimeConfig
|
||||
} from "@ccr/core/mcp/toolhub-config.ts";
|
||||
|
||||
test("ToolHub runtime candidates include the clean Core test build", () => {
|
||||
assert.ok(bundledToolHubMcpEntryPathCandidates().includes(
|
||||
path.join(process.cwd(), ".test-dist", "core", "runtime", "toolhub-mcp.js")
|
||||
));
|
||||
});
|
||||
|
||||
test("ToolHub runtime includes the built-in browser automation backend", () => {
|
||||
const config = createDefaultAppConfig({ generatedConfigFile: "/tmp/ccr-gateway.config.json" });
|
||||
config.toolHub = {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { appendFileSync, existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { Readable } from "node:stream";
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
applyRawTraceRequestLogPolicy,
|
||||
buildRawTraceConfig,
|
||||
createBodySampler,
|
||||
readRawTraceRequestLogBundle,
|
||||
RawTraceSynchronizer
|
||||
} from "@ccr/core/observability/raw-trace-sync.ts";
|
||||
|
||||
@@ -26,7 +27,7 @@ test("raw trace applies metadata-only body privacy while retaining original size
|
||||
});
|
||||
|
||||
assert.equal(policy.action, "enqueue");
|
||||
assert.equal(policy.captureBodies, false);
|
||||
assert.equal(policy.bodyDisposition, "suppress");
|
||||
assert.equal(policy.update.requestBodyText, "");
|
||||
assert.equal(policy.update.requestBodySizeBytes, 1_024);
|
||||
assert.equal(policy.update.requestBodyTruncated, true);
|
||||
@@ -35,15 +36,17 @@ test("raw trace applies metadata-only body privacy while retaining original size
|
||||
assert.equal(policy.update.responseBodyTruncated, true);
|
||||
});
|
||||
|
||||
test("raw trace uses the same successful-request sampling policy as normal request logs", () => {
|
||||
test("raw trace defers successful-request sampling to the final request admission", () => {
|
||||
const config = createConfig();
|
||||
config.observability.requestLogSuccessSampleRate = 0;
|
||||
|
||||
assert.deepEqual(applyRawTraceRequestLogPolicy(config, {
|
||||
const provisional = applyRawTraceRequestLogPolicy(config, {
|
||||
requestBodySizeBytes: 128,
|
||||
requestId: "sampled-request",
|
||||
statusCode: 200
|
||||
}), { action: "discard", reason: "sampled" });
|
||||
});
|
||||
assert.equal(provisional.action, "enqueue");
|
||||
assert.equal(provisional.update.deferOutcomeUntilRecord, true);
|
||||
|
||||
const errorPolicy = applyRawTraceRequestLogPolicy(config, {
|
||||
requestBodySizeBytes: 128,
|
||||
@@ -51,9 +54,29 @@ test("raw trace uses the same successful-request sampling policy as normal reque
|
||||
statusCode: 429
|
||||
});
|
||||
assert.equal(errorPolicy.action, "enqueue");
|
||||
assert.equal(errorPolicy.update.deferOutcomeUntilRecord, true);
|
||||
});
|
||||
|
||||
test("raw trace captures complete error bodies but suppresses successful bodies in errors-only mode", () => {
|
||||
test("raw trace defers HTTP 200 stream sampling and errors-only body policy", () => {
|
||||
const config = createConfig();
|
||||
config.observability.requestLogBodyCapture = "errors";
|
||||
config.observability.requestLogSuccessSampleRate = 0;
|
||||
|
||||
const policy = applyRawTraceRequestLogPolicy(config, {
|
||||
isStream: true,
|
||||
requestId: "stream-with-late-error",
|
||||
responseBodyText: "event: error\ndata: {\"error\":\"late\"}\n\n",
|
||||
statusCode: 200
|
||||
});
|
||||
|
||||
assert.equal(policy.action, "enqueue");
|
||||
assert.equal(policy.bodyDisposition, "defer");
|
||||
assert.equal(policy.update.deferBodyCaptureUntilRecord, true);
|
||||
assert.equal(policy.update.deferOutcomeUntilRecord, true);
|
||||
assert.match(policy.update.responseBodyText, /late/);
|
||||
});
|
||||
|
||||
test("raw trace captures known error bodies and defers provisional 2xx bodies in errors-only mode", () => {
|
||||
const config = createConfig();
|
||||
config.observability.requestLogBodyCapture = "errors";
|
||||
|
||||
@@ -63,7 +86,8 @@ test("raw trace captures complete error bodies but suppresses successful bodies
|
||||
statusCode: 200
|
||||
});
|
||||
assert.equal(successful.action, "enqueue");
|
||||
assert.equal(successful.captureBodies, false);
|
||||
assert.equal(successful.bodyDisposition, "defer");
|
||||
assert.equal(successful.update.deferOutcomeUntilRecord, true);
|
||||
|
||||
const failed = applyRawTraceRequestLogPolicy(config, {
|
||||
requestBodySizeBytes: 64,
|
||||
@@ -71,7 +95,25 @@ test("raw trace captures complete error bodies but suppresses successful bodies
|
||||
statusCode: 500
|
||||
});
|
||||
assert.equal(failed.action, "enqueue");
|
||||
assert.equal(failed.captureBodies, true);
|
||||
assert.equal(failed.bodyDisposition, "capture");
|
||||
});
|
||||
|
||||
test("raw trace defers body persistence when the upstream status is unknown", () => {
|
||||
const config = createConfig();
|
||||
config.observability.requestLogBodyCapture = "errors";
|
||||
config.observability.requestLogSuccessSampleRate = 0;
|
||||
|
||||
const policy = applyRawTraceRequestLogPolicy(config, {
|
||||
requestBodySizeBytes: 64,
|
||||
requestBodyText: "private request body",
|
||||
requestId: "network-failure-without-status"
|
||||
});
|
||||
|
||||
assert.equal(policy.action, "enqueue");
|
||||
assert.equal(policy.bodyDisposition, "defer");
|
||||
assert.equal(policy.update.bodyCapturePolicy, "errors");
|
||||
assert.equal(policy.update.deferBodyCaptureUntilRecord, true);
|
||||
assert.equal(policy.update.requestBodyText, "private request body");
|
||||
});
|
||||
|
||||
test("raw trace source defaults to the 50 MB hard body ceiling", () => {
|
||||
@@ -115,15 +157,19 @@ test("stream sampler retains the original response byte size after capture trunc
|
||||
assert.equal(sampler.sizeBytes(), body.byteLength);
|
||||
});
|
||||
|
||||
test("raw trace sync retains its bundle when the asynchronous queue rejects it", async () => {
|
||||
test("raw trace sync acknowledges only after durable inbox ownership and retains queue rejections", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-sync-reject-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "bundle");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"keep me"}');
|
||||
let enqueueCalls = 0;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async () => false,
|
||||
enqueueUpdate: async () => {
|
||||
enqueueCalls += 1;
|
||||
return false;
|
||||
},
|
||||
getConfig: createConfig,
|
||||
spoolDirectory
|
||||
});
|
||||
@@ -134,6 +180,7 @@ test("raw trace sync retains its bundle when the asynchronous queue rejects it",
|
||||
originalBytes: 21,
|
||||
partType: "upstream_request"
|
||||
}],
|
||||
requestId: "core-bundle-queue-rejected",
|
||||
turnKey: "queue-rejected-request"
|
||||
})]);
|
||||
request.method = "POST";
|
||||
@@ -150,9 +197,678 @@ test("raw trace sync retains its bundle when the asynchronous queue rejects it",
|
||||
|
||||
try {
|
||||
await synchronizer.handle(request, response);
|
||||
assert.equal(result.statusCode, 503);
|
||||
assert.equal(result.body.accepted, false);
|
||||
assert.equal(result.statusCode, 202);
|
||||
assert.equal(result.body.accepted, true);
|
||||
assert.equal(result.body.durable, true);
|
||||
assert.equal(result.body.bundleId, "core-bundle-queue-rejected");
|
||||
assert.equal(existsSync(bundleDirectory), false);
|
||||
const inbox = path.join(spoolDirectory, ".ccr-inbox");
|
||||
assert.equal(readdirSync(inbox).length, 1);
|
||||
assert.equal(existsSync(path.join(inbox, readdirSync(inbox)[0], "upstream_request.json")), true);
|
||||
await waitFor(() => enqueueCalls === 1);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace sync acknowledges and cleans bundles for terminally dropped records", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-sync-record-dropped-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "bundle");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"discard me"}');
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async () => ({
|
||||
accepted: false,
|
||||
degraded: false,
|
||||
reason: "record_dropped"
|
||||
}),
|
||||
getConfig: createConfig,
|
||||
spoolDirectory
|
||||
});
|
||||
const request = Readable.from([JSON.stringify({
|
||||
parts: [{
|
||||
contentType: "application/json",
|
||||
filePath: bodyFile,
|
||||
originalBytes: 24,
|
||||
partType: "upstream_request"
|
||||
}],
|
||||
requestId: "core-bundle-terminal-drop",
|
||||
turnKey: "terminally-dropped-request"
|
||||
})]);
|
||||
request.method = "POST";
|
||||
request.headers = { [rawTraceSyncHeader]: synchronizer.token };
|
||||
const result = {};
|
||||
const response = {
|
||||
end(body) {
|
||||
result.body = JSON.parse(body);
|
||||
},
|
||||
writeHead(statusCode) {
|
||||
result.statusCode = statusCode;
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
await synchronizer.handle(request, response);
|
||||
assert.equal(result.statusCode, 202);
|
||||
assert.equal(result.body.durable, true);
|
||||
await waitFor(() => readdirSync(path.join(spoolDirectory, ".ccr-inbox")).length === 0);
|
||||
assert.equal(existsSync(bundleDirectory), false);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace startup replay outlives producer retries and applies a pending bundle later", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-replay-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "producer-bundle");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"replay me"}');
|
||||
writeFileSync(path.join(bundleDirectory, "manifest.json"), JSON.stringify({
|
||||
parts: [{
|
||||
contentType: "application/json",
|
||||
filePath: bodyFile,
|
||||
originalBytes: 23,
|
||||
partType: "upstream_request"
|
||||
}],
|
||||
requestId: "core-bundle-replay",
|
||||
turnKey: "logical-replay-request",
|
||||
uploadedAt: new Date().toISOString()
|
||||
}));
|
||||
let attempts = 0;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async (_update, files) => {
|
||||
attempts += 1;
|
||||
if (attempts === 1) {
|
||||
return { accepted: false, degraded: false, reason: "record_pending" };
|
||||
}
|
||||
rmSync(files.cleanupDirectory, { force: true, recursive: true });
|
||||
return { accepted: true, degraded: false };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
replayIntervalMs: 10,
|
||||
retryCooldownMs: 10,
|
||||
spoolDirectory
|
||||
});
|
||||
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => attempts >= 2 &&
|
||||
readdirSync(path.join(spoolDirectory, ".ccr-inbox")).length === 0);
|
||||
assert.equal(existsSync(bundleDirectory), false);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace replay recovers an inbox bundle after the accepting process crashes", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-crash-replay-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "producer-bundle");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"survive ack"}');
|
||||
const first = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async () => ({ accepted: true, degraded: false }),
|
||||
getConfig: createConfig,
|
||||
spoolDirectory
|
||||
});
|
||||
const request = Readable.from([JSON.stringify({
|
||||
parts: [{ filePath: bodyFile, originalBytes: 25, partType: "upstream_request" }],
|
||||
requestId: "core-bundle-crash",
|
||||
turnKey: "logical-crash-request"
|
||||
})]);
|
||||
request.method = "POST";
|
||||
request.headers = { [rawTraceSyncHeader]: first.token };
|
||||
const response = { end() {}, writeHead() {} };
|
||||
|
||||
try {
|
||||
await first.handle(request, response);
|
||||
const inbox = path.join(spoolDirectory, ".ccr-inbox");
|
||||
await waitFor(() => readdirSync(inbox).length === 1);
|
||||
|
||||
let replayed = 0;
|
||||
const second = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async (_update, files) => {
|
||||
replayed += 1;
|
||||
rmSync(files.cleanupDirectory, { force: true, recursive: true });
|
||||
return { accepted: true, degraded: false };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
replayIntervalMs: 10,
|
||||
retryCooldownMs: 10,
|
||||
spoolDirectory
|
||||
});
|
||||
await second.start();
|
||||
await waitFor(() => replayed === 1 && readdirSync(inbox).length === 0);
|
||||
await second.stop();
|
||||
} finally {
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace durable ACK waits for every part fsync and refuses ACK when fsync fails", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-fsync-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
try {
|
||||
for (const shouldFail of [false, true]) {
|
||||
const bundleDirectory = path.join(spoolDirectory, `bundle-${shouldFail}`);
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"fsync me"}');
|
||||
let partSynced = false;
|
||||
let responseObservedSync = false;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async () => ({ accepted: false, degraded: false, reason: "record_dropped" }),
|
||||
getConfig: createConfig,
|
||||
spoolDirectory,
|
||||
syncPartFile: async () => {
|
||||
partSynced = true;
|
||||
if (shouldFail) throw new Error("injected fsync failure");
|
||||
}
|
||||
});
|
||||
const result = await sendRawTrace(synchronizer, {
|
||||
parts: [{
|
||||
filePath: bodyFile,
|
||||
partType: "upstream_request",
|
||||
storedBytes: Buffer.byteLength('{"message":"fsync me"}')
|
||||
}],
|
||||
requestId: `fsync-bundle-${shouldFail}`,
|
||||
turnKey: `fsync-request-${shouldFail}`
|
||||
}, () => {
|
||||
responseObservedSync = partSynced;
|
||||
});
|
||||
assert.equal(partSynced, true);
|
||||
assert.equal(responseObservedSync, true);
|
||||
assert.equal(result.statusCode, shouldFail ? 503 : 202);
|
||||
assert.equal(result.body.durable, shouldFail ? undefined : true);
|
||||
await synchronizer.stop();
|
||||
}
|
||||
} finally {
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace publishes a fully durable staging directory atomically", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-atomic-publish-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "bundle");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"atomic"}');
|
||||
let releaseSync;
|
||||
const syncGate = new Promise((resolve) => {
|
||||
releaseSync = resolve;
|
||||
});
|
||||
let syncing = false;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async () => ({ accepted: false, degraded: false, reason: "record_pending" }),
|
||||
getConfig: createConfig,
|
||||
spoolDirectory,
|
||||
syncPartFile: async () => {
|
||||
syncing = true;
|
||||
await syncGate;
|
||||
}
|
||||
});
|
||||
try {
|
||||
const upload = sendRawTrace(synchronizer, {
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: "atomic-publish-bundle",
|
||||
turnKey: "atomic-publish-request"
|
||||
});
|
||||
await waitFor(() => syncing);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-inbox")).length, 0);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-staging")).length, 1);
|
||||
releaseSync();
|
||||
const result = await upload;
|
||||
assert.equal(result.statusCode, 202);
|
||||
const published = readdirSync(path.join(spoolDirectory, ".ccr-inbox"));
|
||||
assert.equal(published.length, 1);
|
||||
assert.equal(existsSync(path.join(spoolDirectory, ".ccr-inbox", published[0], ".ccr-delivery.json")), true);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-staging")).length, 0);
|
||||
} finally {
|
||||
releaseSync?.();
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace startup does not wait for backlog delivery", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-background-replay-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const inboxDirectory = path.join(spoolDirectory, ".ccr-inbox");
|
||||
const bundleDirectory = path.join(inboxDirectory, "backlog");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"backlog"}');
|
||||
writeFileSync(path.join(bundleDirectory, "manifest.json"), JSON.stringify({
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: "background-replay-bundle",
|
||||
turnKey: "background-replay-request"
|
||||
}));
|
||||
let releaseDelivery;
|
||||
const deliveryGate = new Promise((resolve) => {
|
||||
releaseDelivery = resolve;
|
||||
});
|
||||
let deliveryStarted = false;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async () => {
|
||||
deliveryStarted = true;
|
||||
await deliveryGate;
|
||||
return { accepted: false, degraded: false, reason: "record_pending" };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
replayIntervalMs: 60_000,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => deliveryStarted);
|
||||
assert.equal(deliveryStarted, true);
|
||||
} finally {
|
||||
releaseDelivery?.();
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace inbox and dead letters stay within configured capacity", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-capacity-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
deadLetterMaxBundles: 1,
|
||||
deadLetterMaxBytes: 1024 * 1024,
|
||||
enqueueUpdate: async () => ({ accepted: false, degraded: false, reason: "record_pending" }),
|
||||
getConfig: createConfig,
|
||||
inboxMaxBundles: 1,
|
||||
inboxMaxBytes: 1024 * 1024,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
for (let index = 1; index <= 3; index += 1) {
|
||||
const bundleDirectory = path.join(spoolDirectory, `bundle-${index}`);
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, JSON.stringify({ index }));
|
||||
await sendRawTrace(synchronizer, {
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: `capacity-bundle-${index}`,
|
||||
turnKey: `capacity-request-${index}`
|
||||
});
|
||||
await synchronizer.stop();
|
||||
}
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-inbox")).length, 1);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-dead-letter")).length, 1);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace isolates incomplete source bundles under the same bounded retention", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-incomplete-source-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
for (let index = 0; index < 3; index += 1) {
|
||||
const bundleDirectory = path.join(spoolDirectory, `incomplete-${index}`);
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(path.join(bundleDirectory, "orphaned.part"), "x".repeat(128));
|
||||
if (index === 1) writeFileSync(path.join(bundleDirectory, "manifest.json"), "{broken-json");
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
deadLetterMaxBundles: 1,
|
||||
deadLetterMaxBytes: 1024 * 1024,
|
||||
getConfig: createConfig,
|
||||
inboxMaxBundles: 1,
|
||||
inboxMaxBytes: 1024 * 1024,
|
||||
replayIntervalMs: 5,
|
||||
sourceBundleGraceMs: 1,
|
||||
sourceScanIntervalMs: 1,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => readdirSync(spoolDirectory)
|
||||
.filter((name) => ![".ccr-inbox", ".ccr-dead-letter", ".ccr-staging"].includes(name))
|
||||
.length === 0);
|
||||
const sourceEntries = readdirSync(spoolDirectory)
|
||||
.filter((name) => ![".ccr-inbox", ".ccr-dead-letter", ".ccr-staging"].includes(name));
|
||||
assert.deepEqual(sourceEntries, []);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-inbox")).length, 0);
|
||||
await waitFor(() => readdirSync(path.join(spoolDirectory, ".ccr-dead-letter")).length === 1);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-dead-letter")).length, 1);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace keeps an over-capacity streaming source while its files are active", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-active-source-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "active-stream");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_response.sse");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, "data: start\n\n");
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
getConfig: createConfig,
|
||||
inboxMaxBytes: 1,
|
||||
replayIntervalMs: 5,
|
||||
sourceBundleGraceMs: 30,
|
||||
sourceScanIntervalMs: 5,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
for (let index = 0; index < 4; index += 1) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 15));
|
||||
appendFileSync(bodyFile, `data: ${index}\n\n`);
|
||||
}
|
||||
assert.equal(existsSync(bundleDirectory), true);
|
||||
await waitFor(() => !existsSync(bundleDirectory));
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-dead-letter")).length, 1);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace ACK measures only the newly admitted bundle after startup indexing", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-incremental-capacity-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const inboxDirectory = path.join(spoolDirectory, ".ccr-inbox");
|
||||
mkdirSync(inboxDirectory, { recursive: true });
|
||||
for (let index = 0; index < 40; index += 1) {
|
||||
const bundleDirectory = path.join(inboxDirectory, `backlog-${index}`);
|
||||
const metadataFile = path.join(bundleDirectory, "upstream_response_metadata.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(metadataFile, JSON.stringify({ statusCode: 200 }));
|
||||
writeFileSync(path.join(bundleDirectory, "manifest.json"), JSON.stringify({
|
||||
parts: [{ filePath: metadataFile, partType: "upstream_response_metadata" }],
|
||||
requestId: `backlog-bundle-${index}`,
|
||||
turnKey: `backlog-request-${index}`
|
||||
}));
|
||||
}
|
||||
const measuredDirectories = [];
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async () => ({ accepted: false, degraded: false, reason: "record_pending" }),
|
||||
getConfig: createConfig,
|
||||
measureDirectorySize: async (directory) => {
|
||||
measuredDirectories.push(directory);
|
||||
return 1;
|
||||
},
|
||||
replayIntervalMs: 60_000,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => measuredDirectories.length === 40);
|
||||
assert.equal(measuredDirectories.length, 40);
|
||||
measuredDirectories.length = 0;
|
||||
|
||||
const producerDirectory = path.join(spoolDirectory, "new-producer-bundle");
|
||||
const bodyFile = path.join(producerDirectory, "upstream_request.json");
|
||||
mkdirSync(producerDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"incremental"}');
|
||||
let result;
|
||||
const deadline = Date.now() + 2_000;
|
||||
do {
|
||||
result = await sendRawTrace(synchronizer, {
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: "incremental-capacity-bundle",
|
||||
turnKey: "incremental-capacity-request"
|
||||
});
|
||||
if (result.statusCode !== 503) break;
|
||||
assert.equal(result.body.reason, "initializing");
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
} while (Date.now() < deadline);
|
||||
assert.equal(result.statusCode, 202);
|
||||
assert.equal(measuredDirectories.length, 1);
|
||||
assert.equal(measuredDirectories[0].includes("new-producer-bundle"), false);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace moves permanently pending bundles to bounded dead letter after max attempts", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-max-attempt-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "pending-bundle");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"pending"}');
|
||||
writeFileSync(path.join(bundleDirectory, "manifest.json"), JSON.stringify({
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: "max-attempt-bundle",
|
||||
turnKey: "max-attempt-request"
|
||||
}));
|
||||
let attempts = 0;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
bundleMaxAttempts: 1,
|
||||
enqueueUpdate: async () => {
|
||||
attempts += 1;
|
||||
return { accepted: false, degraded: false, reason: "record_pending" };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
replayIntervalMs: 10,
|
||||
retryCooldownMs: 10,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => readdirSync(path.join(spoolDirectory, ".ccr-dead-letter")).length === 1);
|
||||
assert.equal(attempts, 1);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-inbox")).length, 0);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace backs off record-pending retries without repeatedly persisting attempts", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-pending-backoff-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const bundleDirectory = path.join(spoolDirectory, "pending-backoff-bundle");
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, '{"message":"pending"}');
|
||||
let attempts = 0;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
bundleMaxAttempts: 100,
|
||||
enqueueUpdate: async () => {
|
||||
attempts += 1;
|
||||
return { accepted: false, degraded: false, reason: "record_pending" };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
pendingRetryMaxMs: 1_000,
|
||||
replayIntervalMs: 5,
|
||||
retryCooldownMs: 15,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
const result = await sendRawTrace(synchronizer, {
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: "pending-backoff-bundle-id",
|
||||
turnKey: "pending-backoff-request"
|
||||
});
|
||||
assert.equal(result.statusCode, 202);
|
||||
await synchronizer.start();
|
||||
await waitFor(() => attempts >= 1);
|
||||
const inboxDirectory = path.join(spoolDirectory, ".ccr-inbox");
|
||||
const storedDirectory = path.join(inboxDirectory, readdirSync(inboxDirectory)[0]);
|
||||
const deliveryFile = path.join(storedDirectory, ".ccr-delivery.json");
|
||||
await waitFor(() => JSON.parse(readFileSync(deliveryFile, "utf8")).lastError === "record_pending");
|
||||
const firstPersistedState = readFileSync(deliveryFile, "utf8");
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 80));
|
||||
assert.ok(attempts >= 2);
|
||||
assert.ok(attempts <= 3);
|
||||
assert.equal(readFileSync(deliveryFile, "utf8"), firstPersistedState);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace replay rotates through a bounded number of bundles per pass", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-replay-budget-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const inboxDirectory = path.join(spoolDirectory, ".ccr-inbox");
|
||||
mkdirSync(inboxDirectory, { recursive: true });
|
||||
for (let index = 0; index < 5; index += 1) {
|
||||
const bundleDirectory = path.join(inboxDirectory, `budget-${index}`);
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, JSON.stringify({ index }));
|
||||
writeFileSync(path.join(bundleDirectory, "manifest.json"), JSON.stringify({
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: `budget-bundle-${index}`,
|
||||
turnKey: `budget-request-${index}`
|
||||
}));
|
||||
}
|
||||
const attemptedBundleIds = [];
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
bundleMaxAttempts: 100,
|
||||
enqueueUpdate: async (update) => {
|
||||
attemptedBundleIds.push(update.bundleId);
|
||||
return { accepted: false, degraded: false, reason: "record_pending" };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
replayIntervalMs: 50,
|
||||
replayMaxBundlesPerPass: 2,
|
||||
replayTimeBudgetMs: 1_000,
|
||||
retryCooldownMs: 60_000,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => attemptedBundleIds.length >= 2);
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
assert.equal(attemptedBundleIds.length, 2);
|
||||
await waitFor(() => attemptedBundleIds.length === 5);
|
||||
assert.equal(new Set(attemptedBundleIds).size, 5);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace replay stops when its time budget is exhausted", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-replay-time-budget-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
const inboxDirectory = path.join(spoolDirectory, ".ccr-inbox");
|
||||
mkdirSync(inboxDirectory, { recursive: true });
|
||||
for (let index = 0; index < 3; index += 1) {
|
||||
const bundleDirectory = path.join(inboxDirectory, `timed-${index}`);
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, JSON.stringify({ index }));
|
||||
writeFileSync(path.join(bundleDirectory, "manifest.json"), JSON.stringify({
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: `timed-bundle-${index}`,
|
||||
turnKey: `timed-request-${index}`
|
||||
}));
|
||||
}
|
||||
let attempts = 0;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
bundleMaxAttempts: 100,
|
||||
enqueueUpdate: async () => {
|
||||
attempts += 1;
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
return { accepted: false, degraded: false, reason: "record_pending" };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
replayIntervalMs: 200,
|
||||
replayMaxBundlesPerPass: 100,
|
||||
replayTimeBudgetMs: 5,
|
||||
retryCooldownMs: 60_000,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => attempts >= 1);
|
||||
await new Promise((resolve) => setTimeout(resolve, 60));
|
||||
assert.equal(attempts, 1);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("raw trace replay isolates a failing bundle and continues with later bundles", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-isolated-replay-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
for (const name of ["bad", "good"]) {
|
||||
const bundleDirectory = path.join(spoolDirectory, `${name}-bundle`);
|
||||
const bodyFile = path.join(bundleDirectory, "upstream_request.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(bodyFile, JSON.stringify({ name }));
|
||||
writeFileSync(path.join(bundleDirectory, "manifest.json"), JSON.stringify({
|
||||
parts: [{ filePath: bodyFile, partType: "upstream_request" }],
|
||||
requestId: `${name}-bundle-id`,
|
||||
turnKey: `${name}-request`
|
||||
}));
|
||||
}
|
||||
let goodApplied = false;
|
||||
const synchronizer = new RawTraceSynchronizer({
|
||||
enqueueUpdate: async (update, files) => {
|
||||
if (update.bundleId === "bad-bundle-id") throw new Error("permanent bad bundle");
|
||||
goodApplied = true;
|
||||
rmSync(files.cleanupDirectory, { force: true, recursive: true });
|
||||
return { accepted: true, degraded: false };
|
||||
},
|
||||
getConfig: createConfig,
|
||||
replayIntervalMs: 10_000,
|
||||
spoolDirectory
|
||||
});
|
||||
try {
|
||||
await synchronizer.start();
|
||||
await waitFor(() => goodApplied);
|
||||
assert.equal(goodApplied, true);
|
||||
assert.equal(readdirSync(path.join(spoolDirectory, ".ccr-inbox")).length, 1);
|
||||
} finally {
|
||||
await synchronizer.stop();
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("fallback raw bundles keep unique bundle ids while sharing the logical request", async () => {
|
||||
const dir = mkdtempSync(path.join(tmpdir(), "ccr-raw-trace-fallback-id-test-"));
|
||||
const spoolDirectory = path.join(dir, "spool");
|
||||
try {
|
||||
const bundles = [];
|
||||
for (const attempt of [1, 2]) {
|
||||
const bundleDirectory = path.join(spoolDirectory, `bundle-${attempt}`);
|
||||
const clientMetadata = path.join(bundleDirectory, "client_request_metadata.json");
|
||||
const responseMetadata = path.join(bundleDirectory, "upstream_response_metadata.json");
|
||||
mkdirSync(bundleDirectory, { recursive: true });
|
||||
writeFileSync(clientMetadata, JSON.stringify({ headers: { "x-ccr-route-attempt": String(attempt) } }));
|
||||
writeFileSync(responseMetadata, JSON.stringify({ statusCode: attempt === 1 ? 500 : 200 }));
|
||||
bundles.push(await readRawTraceRequestLogBundle({
|
||||
parts: [
|
||||
{ filePath: clientMetadata, partType: "client_request_metadata" },
|
||||
{ filePath: responseMetadata, partType: "upstream_response_metadata" }
|
||||
],
|
||||
requestId: `core-bundle-${attempt}`,
|
||||
turnKey: "shared-logical-request"
|
||||
}, spoolDirectory));
|
||||
}
|
||||
|
||||
assert.deepEqual(bundles.map((bundle) => ({
|
||||
attempt: bundle.update.attempt,
|
||||
bundleId: bundle.update.bundleId,
|
||||
requestId: bundle.update.requestId
|
||||
})), [
|
||||
{ attempt: 1, bundleId: "core-bundle-1", requestId: "shared-logical-request" },
|
||||
{ attempt: 2, bundleId: "core-bundle-2", requestId: "shared-logical-request" }
|
||||
]);
|
||||
} finally {
|
||||
rmSync(dir, { force: true, recursive: true });
|
||||
}
|
||||
@@ -165,3 +881,29 @@ function createConfig() {
|
||||
config.observability.requestLogSuccessSampleRate = 1;
|
||||
return config;
|
||||
}
|
||||
|
||||
async function waitFor(predicate, timeoutMs = 2_000) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (!predicate()) {
|
||||
if (Date.now() >= deadline) assert.fail("Timed out waiting for raw trace state.");
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
}
|
||||
}
|
||||
|
||||
async function sendRawTrace(synchronizer, manifest, onResponse) {
|
||||
const request = Readable.from([JSON.stringify(manifest)]);
|
||||
request.method = "POST";
|
||||
request.headers = { [rawTraceSyncHeader]: synchronizer.token };
|
||||
const result = {};
|
||||
const response = {
|
||||
end(body) {
|
||||
result.body = JSON.parse(body);
|
||||
onResponse?.();
|
||||
},
|
||||
writeHead(statusCode) {
|
||||
result.statusCode = statusCode;
|
||||
}
|
||||
};
|
||||
await synchronizer.handle(request, response);
|
||||
return result;
|
||||
}
|
||||
|
||||
@@ -1,6 +1,23 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
import { RequestRouteTraceRecorder } from "@ccr/core/observability/route-trace.ts";
|
||||
import { isSensitiveRequestLogHeaderName } from "@ccr/core/observability/sensitive-headers.ts";
|
||||
|
||||
test("request log header redaction fails closed for custom authentication headers", () => {
|
||||
for (const name of [
|
||||
"x-auth-token",
|
||||
"x-amz-security-token",
|
||||
"x-company-client-secret",
|
||||
"x-private-key",
|
||||
"x-signed-request-signature",
|
||||
"x-custom-credential"
|
||||
]) {
|
||||
assert.equal(isSensitiveRequestLogHeaderName(name), true, name);
|
||||
}
|
||||
for (const name of ["content-type", "user-agent", "x-request-id"]) {
|
||||
assert.equal(isSensitiveRequestLogHeaderName(name), false, name);
|
||||
}
|
||||
});
|
||||
|
||||
test("route trace records actively reported changes and never persists sensitive values", () => {
|
||||
const startedAt = Date.now();
|
||||
@@ -11,6 +28,7 @@ test("route trace records actively reported changes and never persists sensitive
|
||||
{ after: "model-b", before: "model-a", operation: "replace", path: "/body/model", scope: "body" },
|
||||
{ after: "body-secret-b", before: "body-secret-a", operation: "replace", path: "/body/api_key", scope: "body" },
|
||||
{ after: "Bearer header-secret-b", before: "Bearer header-secret-a", operation: "replace", path: "/headers/authorization", scope: "headers" },
|
||||
{ after: "auth-sub-secret-b", before: "auth-sub-secret-a", operation: "replace", path: "/headers/x-auth-sub", scope: "headers" },
|
||||
{
|
||||
after: "https://upstream.example/v1/messages?access_token=url-secret-b",
|
||||
before: "http://127.0.0.1/v1/messages?access_token=url-secret-a",
|
||||
@@ -36,7 +54,42 @@ test("route trace records actively reported changes and never persists sensitive
|
||||
assert.ok(trace.hops[1].changes.some((change) => change.path === "/body/model"));
|
||||
assert.ok(trace.hops[1].changes.some((change) => change.path === "/headers/authorization" && change.redacted));
|
||||
assert.match(serialized, /\[redacted\]/);
|
||||
assert.doesNotMatch(serialized, /header-secret|body-secret|url-secret/);
|
||||
assert.ok(trace.hops[1].changes.some((change) => change.path === "/headers/x-auth-sub" && change.redacted));
|
||||
assert.doesNotMatch(serialized, /header-secret|auth-sub-secret|body-secret|url-secret/);
|
||||
});
|
||||
|
||||
test("route trace omits body values when request body capture is disabled", () => {
|
||||
const recorder = new RequestRouteTraceRecorder(Date.now());
|
||||
recorder.captureIngress();
|
||||
recorder.capture({
|
||||
changes: [
|
||||
{
|
||||
after: [{ role: "user", content: "after-private-message" }],
|
||||
before: [{ role: "user", content: "before-private-message" }],
|
||||
operation: "replace",
|
||||
path: "/body/messages",
|
||||
scope: "body"
|
||||
},
|
||||
{
|
||||
after: "diagnostic-value",
|
||||
operation: "add",
|
||||
path: "/headers/x-ccr-route-source",
|
||||
scope: "headers"
|
||||
}
|
||||
],
|
||||
name: "router.policy",
|
||||
phase: "routing"
|
||||
});
|
||||
|
||||
const trace = recorder.finish({ captureBodyValues: false });
|
||||
const bodyChange = trace.hops[1].changes[0];
|
||||
assert.deepEqual(bodyChange, {
|
||||
operation: "replace",
|
||||
path: "/body/messages",
|
||||
scope: "body"
|
||||
});
|
||||
assert.equal(trace.hops[1].changes[1].after, "diagnostic-value");
|
||||
assert.doesNotMatch(JSON.stringify(trace), /private-message/);
|
||||
});
|
||||
|
||||
test("route trace bounds actively reported values without parsing request bodies", () => {
|
||||
|
||||
Reference in New Issue
Block a user