feat: add code queue services and baidu netdisk
This commit is contained in:
Codex
@@ -1,7 +1,6 @@
|
||||
import { appendFileSync, mkdirSync } from "node:fs";
|
||||
import { dirname } from "node:path";
|
||||
import type { Server, ServerWebSocket } from "bun";
|
||||
import postgres from "postgres";
|
||||
import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../shared/src/rotating-jsonl";
|
||||
import {
|
||||
type ApiEvent,
|
||||
type ApiNode,
|
||||
@@ -145,6 +144,21 @@ const microserviceProxyRefreshes = new Map<string, Promise<void>>();
|
||||
let lastTaskSweepAt = 0;
|
||||
let taskSweepInFlight: Promise<void> | null = null;
|
||||
const microserviceProxyMaxBodyTextLength = 8 * 1024 * 1024;
|
||||
const microserviceForwardRequestHeaders = [
|
||||
"accept",
|
||||
"content-type",
|
||||
"range",
|
||||
"x-auth",
|
||||
"x-requested-with",
|
||||
"destination",
|
||||
"overwrite",
|
||||
"tus-resumable",
|
||||
"upload-concat",
|
||||
"upload-defer-length",
|
||||
"upload-length",
|
||||
"upload-metadata",
|
||||
"upload-offset",
|
||||
] as const;
|
||||
|
||||
function requiredEnv(name: string): string {
|
||||
const value = process.env[name];
|
||||
@@ -272,7 +286,12 @@ function readConfig(): RuntimeConfig {
|
||||
}
|
||||
|
||||
function createLogger(service: string, logFile: string) {
|
||||
mkdirSync(dirname(logFile), { recursive: true });
|
||||
const writer = createHourlyJsonlWriter({
|
||||
baseLogFile: logFile,
|
||||
service,
|
||||
maxBytes: logRetentionBytesForService(service),
|
||||
});
|
||||
writer.prune();
|
||||
return (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue): void => {
|
||||
const entry = data === undefined
|
||||
? { ts: new Date().toISOString(), service, level, message }
|
||||
@@ -281,7 +300,7 @@ function createLogger(service: string, logFile: string) {
|
||||
while (recentLogs.length > 500) recentLogs.shift();
|
||||
const line = `${JSON.stringify(entry)}\n`;
|
||||
try {
|
||||
appendFileSync(logFile, line, "utf8");
|
||||
writer.appendLine(line, new Date(entry.ts));
|
||||
} catch (error) {
|
||||
console.error(JSON.stringify({ ts: new Date().toISOString(), service, level: "error", message: "log_write_failed", data: String(error) }));
|
||||
}
|
||||
@@ -1515,6 +1534,24 @@ function contentTypeIsJson(contentType: string): boolean {
|
||||
return contentType.toLowerCase().includes("json");
|
||||
}
|
||||
|
||||
function readMicroserviceRequestHeaders(req: Request): Record<string, JsonValue> {
|
||||
const requestHeaders: Record<string, JsonValue> = {};
|
||||
for (const name of microserviceForwardRequestHeaders) {
|
||||
const value = req.headers.get(name);
|
||||
if (value !== null && value.length > 0) requestHeaders[name] = value.slice(0, 4096);
|
||||
}
|
||||
return requestHeaders;
|
||||
}
|
||||
|
||||
function headersFromMicroserviceRequest(requestHeaders: Record<string, JsonValue>): Headers {
|
||||
const headers = new Headers();
|
||||
for (const name of microserviceForwardRequestHeaders) {
|
||||
const value = requestHeaders[name];
|
||||
if (typeof value === "string" && value.length > 0) headers.set(name, value);
|
||||
}
|
||||
return headers;
|
||||
}
|
||||
|
||||
function boundedMicroserviceBodyText(
|
||||
bodyText: string,
|
||||
contentType: string,
|
||||
@@ -1620,7 +1657,7 @@ function microserviceCacheTtlMs(serviceId: string, targetPath: string): number {
|
||||
if (serviceId === "findjob" && (targetPath === "/api/summary" || targetPath === "/api/jobs" || targetPath === "/api/drafts")) return 8_000;
|
||||
if (serviceId === "met-nonlinear" && (targetPath === "/api/images" || targetPath === "/api/projects")) return 15_000;
|
||||
if (serviceId === "met-nonlinear" && (targetPath === "/api/queue" || targetPath === "/api/summary" || targetPath === "/api/history")) return 5_000;
|
||||
if (serviceId === "codex-queue" && targetPath.includes("/transcript")) return 1_000;
|
||||
if (serviceId === "code-queue" && targetPath.includes("/transcript")) return 1_000;
|
||||
return 750;
|
||||
}
|
||||
|
||||
@@ -1728,9 +1765,7 @@ async function directMicroserviceResponse(
|
||||
const baseUrl = new URL(service.backend.nodeBaseUrl);
|
||||
const upstreamUrl = new URL(targetPath, baseUrl);
|
||||
upstreamUrl.search = proxyOptions.query;
|
||||
const headers = new Headers();
|
||||
const contentType = typeof requestHeaders["content-type"] === "string" ? requestHeaders["content-type"] : "";
|
||||
if (contentType.length > 0) headers.set("content-type", contentType);
|
||||
const headers = headersFromMicroserviceRequest(requestHeaders);
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), Math.max(1000, service.backend.timeoutMs));
|
||||
try {
|
||||
@@ -1857,9 +1892,7 @@ async function microserviceRoute(req: Request, url: URL): Promise<Response> {
|
||||
if (bodyText.length > 1024 * 1024) {
|
||||
return jsonResponse({ ok: false, error: "microservice request body is too large", maxBytes: 1024 * 1024 }, 413);
|
||||
}
|
||||
const requestHeaders: Record<string, JsonValue> = {};
|
||||
const contentType = req.headers.get("content-type");
|
||||
if (contentType !== null) requestHeaders["content-type"] = contentType.slice(0, 200);
|
||||
const requestHeaders = readMicroserviceRequestHeaders(req);
|
||||
if (method === "GET" || method === "HEAD") {
|
||||
const stale = readStaleMicroserviceCache(cacheKey);
|
||||
if (stale !== null) {
|
||||
@@ -1920,7 +1953,7 @@ function recordFromJson(value: JsonValue | null): Record<string, unknown> {
|
||||
|
||||
async function runHostSshPerfCommand(providerId: string, command: string, timeoutMs = 18_000): Promise<{ ok: boolean; taskId: string; task: RawTaskRow | null; stdout: string; stderr: string; exitCode: number | null; timedOut: boolean }> {
|
||||
const { taskId, providerOnline } = await createAndSendTask(providerId, "host.ssh", {
|
||||
source: "codex-queue-performance-panel",
|
||||
source: "code-queue-performance-panel",
|
||||
mode: "exec",
|
||||
cwd: "/root/unidesk",
|
||||
timeoutMs: 15_000,
|
||||
@@ -1974,7 +2007,7 @@ function parsePerfJson(body: string): Record<string, unknown> | null {
|
||||
|
||||
async function codexQueueLoadTest(req: Request): Promise<Response> {
|
||||
const body = req.method === "POST" ? (await req.json().catch(() => ({}))) as Record<string, unknown> : {};
|
||||
const codexService = microserviceById("codex-queue");
|
||||
const codexService = microserviceById("code-queue");
|
||||
const providerId = typeof body.providerId === "string" && body.providerId.length > 0
|
||||
? body.providerId
|
||||
: codexService?.providerId ?? "main-server";
|
||||
@@ -1985,7 +2018,7 @@ async function codexQueueLoadTest(req: Request): Promise<Response> {
|
||||
const timeoutMs = numberFromUnknown(body.timeoutMs, 90_000, 5_000, 180_000);
|
||||
const targetMs = numberFromUnknown(body.targetMs, 1_000, 100, 60_000);
|
||||
const runId = safePerfRunId();
|
||||
const dir = ".state/codex-queue-perf";
|
||||
const dir = ".state/code-queue-perf";
|
||||
const browsersPath = ".state/playwright-browsers";
|
||||
const outputPath = `${dir}/${runId}.json`;
|
||||
const stderrPath = `${dir}/${runId}.stderr`;
|
||||
@@ -1999,7 +2032,7 @@ async function codexQueueLoadTest(req: Request): Promise<Response> {
|
||||
const startCommand = [
|
||||
`mkdir -p ${shellQuote(dir)}`,
|
||||
`rm -f ${shellQuote(outputPath)} ${shellQuote(stderrPath)} ${shellQuote(exitPath)}`,
|
||||
`(${playwrightSetup}; PLAYWRIGHT_BROWSERS_PATH=${shellQuote(browsersPath)} bun scripts/src/codex-queue-perf.ts --json --timeout-ms ${timeoutMs} --target-ms ${targetMs}${urlArg} > ${shellQuote(outputPath)}; printf '%s' "$?" > ${shellQuote(exitPath)}) > ${shellQuote(stderrPath)} 2>&1 & printf '%s\\n' ${shellQuote(runId)}`,
|
||||
`(${playwrightSetup}; PLAYWRIGHT_BROWSERS_PATH=${shellQuote(browsersPath)} bun scripts/src/code-queue-perf.ts --json --timeout-ms ${timeoutMs} --target-ms ${targetMs}${urlArg} > ${shellQuote(outputPath)}; printf '%s' "$?" > ${shellQuote(exitPath)}) > ${shellQuote(stderrPath)} 2>&1 & printf '%s\\n' ${shellQuote(runId)}`,
|
||||
].join("; ");
|
||||
const startedAt = Date.now();
|
||||
const start = await runHostSshPerfCommand(providerId, startCommand);
|
||||
@@ -2058,7 +2091,7 @@ async function codexQueueLoadTest(req: Request): Promise<Response> {
|
||||
runId,
|
||||
stage: "timeout",
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
error: `Codex Queue Playwright benchmark did not finish within ${timeoutMs}ms`,
|
||||
error: `Code Queue Playwright benchmark did not finish within ${timeoutMs}ms`,
|
||||
latestTaskId: latestPoll?.taskId ?? start.taskId,
|
||||
latestTask: rawTaskJson(latestPoll?.task ?? start.task),
|
||||
}, 200);
|
||||
@@ -2197,7 +2230,7 @@ async function routeInner(req: Request, server: Server<WsData>): Promise<Respons
|
||||
}
|
||||
if (url.pathname === "/api/microservices") return jsonResponse({ ok: true, microservices: await withPerformanceOperation("core", "microservices", url.pathname, () => getMicroservices()) });
|
||||
if (url.pathname === "/api/performance") return jsonResponse(await getPerformance());
|
||||
if (url.pathname === "/api/codex-queue-load-test" && (req.method === "GET" || req.method === "POST")) return withPerformanceOperation("performance", "codex_queue_load_test", url.pathname, () => codexQueueLoadTest(req));
|
||||
if (url.pathname === "/api/code-queue-load-test" && (req.method === "GET" || req.method === "POST")) return withPerformanceOperation("performance", "code_queue_load_test", url.pathname, () => codexQueueLoadTest(req));
|
||||
if (url.pathname.startsWith("/api/microservices/")) return withPerformanceOperation("microservices", "route", url.pathname, () => microserviceRoute(req, url));
|
||||
if (url.pathname === "/api/dispatch" && req.method === "POST") return withPerformanceOperation("scheduler", "dispatch", url.pathname, () => dispatchTask(req));
|
||||
if (url.pathname === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-readLimit(url, 100)) });
|
||||
|
||||
Reference in New Issue
Block a user