fix(code-queue): stabilize oa publisher and redeploy path
This commit is contained in:
Codex
@@ -116,6 +116,7 @@ import { ReferenceTaskLookupError, configureReferences, injectReferencedTaskCont
|
||||
import {
|
||||
applyOaTraceStatsToTaskJson,
|
||||
configureOaEvents,
|
||||
oaEventPublisherStatus,
|
||||
publishCodeQueueJudgeEvent,
|
||||
publishCodeQueueQueueUpdated,
|
||||
publishCodeQueueTaskUpdated,
|
||||
@@ -3789,6 +3790,7 @@ async function route(req: Request): Promise<Response> {
|
||||
service: "code-queue",
|
||||
queue: await queueSummaryForHealth(false),
|
||||
egressProxy: await providerGatewayEgressProxyStatus(),
|
||||
oaEventPublisher: oaEventPublisherStatus(),
|
||||
startedAt: serviceStartedAt,
|
||||
});
|
||||
if (url.pathname === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-parseLimit(url)) });
|
||||
|
||||
@@ -129,8 +129,8 @@ function fallbackJudge(result: CodexRunResult, minimaxError?: string | JudgeFail
|
||||
}
|
||||
|
||||
export const retryInstruction = "这是同一个 Codex thread 的 continuation，不是新任务。请基于上文继续完成原始任务；只做最小必要状态核查，避免从头重新摸索、避免重复已经完成的修改。";
|
||||
const codeQueueD601DeployCommand = "bun scripts/cli.ts ssh D601 'cd ~/cq-deploy && CODE_QUEUE_ENV_FILE=.state/code-queue-d601.env docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --build --force-recreate code-queue'";
|
||||
const codeQueueRestartSafetyGuidance = `Code Queue 服务具备 restart-recovery，允许在任何时候重启、重建或替换 D601 的 \`code-queue-backend\`；当当前任务修改 Code Queue 自身时，禁止等待当前 Code Queue task 退出或等待队列归零后再重启，因为这会等待自己退出形成自锁。正确做法是在 D601 直接执行 \`${codeQueueD601DeployCommand}\` 或等价 no-deps force-recreate，并在恢复后用 live health/task 查询验证。`;
|
||||
const codeQueueD601DeployCommand = "bun scripts/cli.ts ssh D601 \"nohup bash -lc 'cd ~/cq-deploy && CODE_QUEUE_ENV_FILE=.state/code-queue-d601.env docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --build --force-recreate code-queue > /tmp/unidesk-code-queue-redeploy.log 2>&1' >/tmp/unidesk-code-queue-redeploy-launch.log 2>&1 & echo scheduled:/tmp/unidesk-code-queue-redeploy.log\"";
|
||||
const codeQueueRestartSafetyGuidance = `Code Queue 服务具备 restart-recovery，允许在任何时候重启、重建或替换 D601 的 \`code-queue-backend\`；当当前任务修改 Code Queue 自身时，禁止等待当前 Code Queue task 退出或等待队列归零后再重启，因为这会等待自己退出形成自锁。正确做法是通过 D601 provider SSH 在宿主侧 detached 调度 \`${codeQueueD601DeployCommand}\` 或等价 detached no-deps force-recreate；禁止从正在运行的 \`code-queue-backend\` 容器里以前台方式执行会重建自身的 compose 命令，否则发起命令的容器被替换时可能留下 Created/Exited 的半完成状态。调度后在恢复的 live health/task 查询中验证。`;
|
||||
|
||||
export function explicitUserInterrupt(task: QueueTask, result: CodexRunResult): boolean {
|
||||
return result.terminalStatus === "interrupted"
|
||||
|
||||
@@ -23,6 +23,11 @@ interface OaEventEnvelope {
|
||||
payload: JsonRecord;
|
||||
}
|
||||
|
||||
interface QueuedOaEvent {
|
||||
event: OaEventEnvelope;
|
||||
attempts: number;
|
||||
}
|
||||
|
||||
export interface OaTraceStats extends JsonRecord {
|
||||
scopeId: string;
|
||||
source: "oa-event-flow";
|
||||
@@ -42,8 +47,22 @@ export interface OaTraceStepSummary {
|
||||
source: "oa-event-flow";
|
||||
}
|
||||
|
||||
const postTimeoutMs = 2500;
|
||||
const postTimeoutMs = 10000;
|
||||
const publishConcurrency = 4;
|
||||
const maxQueuedEvents = 20_000;
|
||||
const maxPublishAttempts = 5;
|
||||
let context: OaEventContext | null = null;
|
||||
let queuedEvents: QueuedOaEvent[] = [];
|
||||
let activePublishes = 0;
|
||||
let totalEnqueued = 0;
|
||||
let totalSucceeded = 0;
|
||||
let totalFailed = 0;
|
||||
let totalRetried = 0;
|
||||
let totalAbandoned = 0;
|
||||
let totalDropped = 0;
|
||||
let lastEnqueuedAt: string | null = null;
|
||||
let lastSucceededAt: string | null = null;
|
||||
let lastError: JsonRecord | null = null;
|
||||
|
||||
export function configureOaEvents(runtimeContext: OaEventContext): void {
|
||||
const baseUrl = runtimeContext.baseUrl.replace(/\/+$/u, "");
|
||||
@@ -78,22 +97,91 @@ function taskTags(task: QueueTask, queueId: string, extra: string[] = [], attemp
|
||||
return tags;
|
||||
}
|
||||
|
||||
function postOaEvent(event: OaEventEnvelope): void {
|
||||
export function oaEventPublisherStatus(): JsonRecord {
|
||||
return {
|
||||
pending: queuedEvents.length,
|
||||
active: activePublishes,
|
||||
concurrency: publishConcurrency,
|
||||
maxQueuedEvents,
|
||||
maxPublishAttempts,
|
||||
totalEnqueued,
|
||||
totalSucceeded,
|
||||
totalFailed,
|
||||
totalRetried,
|
||||
totalAbandoned,
|
||||
totalDropped,
|
||||
lastEnqueuedAt,
|
||||
lastSucceededAt,
|
||||
lastError,
|
||||
};
|
||||
}
|
||||
|
||||
async function sendOaEvent(item: QueuedOaEvent): Promise<boolean> {
|
||||
const runtime = ctx();
|
||||
const event = item.event;
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), postTimeoutMs);
|
||||
void fetch(`${runtime.baseUrl}/api/events`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(event),
|
||||
signal: controller.signal,
|
||||
}).then(async (response) => {
|
||||
try {
|
||||
const response = await fetch(`${runtime.baseUrl}/api/events`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(event),
|
||||
signal: controller.signal,
|
||||
});
|
||||
if (!response.ok) {
|
||||
runtime.logger("warn", "oa_event_publish_failed", { eventId: event.eventId, type: event.type, status: response.status, body: safePreview(await response.text(), 1000) });
|
||||
totalFailed += 1;
|
||||
lastError = { eventId: event.eventId, type: event.type, attempt: item.attempts, status: response.status, body: safePreview(await response.text(), 1000) };
|
||||
runtime.logger("warn", "oa_event_publish_failed", lastError);
|
||||
return false;
|
||||
}
|
||||
}).catch((error) => {
|
||||
runtime.logger("warn", "oa_event_publish_error", { eventId: event.eventId, type: event.type, error: error instanceof Error ? error.message : String(error) });
|
||||
}).finally(() => clearTimeout(timer));
|
||||
totalSucceeded += 1;
|
||||
lastSucceededAt = runtime.nowIso();
|
||||
return true;
|
||||
} catch (error) {
|
||||
totalFailed += 1;
|
||||
lastError = { eventId: event.eventId, type: event.type, attempt: item.attempts, error: error instanceof Error ? error.message : String(error) };
|
||||
runtime.logger("warn", "oa_event_publish_error", lastError);
|
||||
return false;
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
function pumpOaEventQueue(): void {
|
||||
while (activePublishes < publishConcurrency && queuedEvents.length > 0) {
|
||||
const item = queuedEvents.shift();
|
||||
if (item === undefined) return;
|
||||
activePublishes += 1;
|
||||
void sendOaEvent(item).then((ok) => {
|
||||
if (!ok) {
|
||||
if (item.attempts < maxPublishAttempts) {
|
||||
totalRetried += 1;
|
||||
queuedEvents.push({ event: item.event, attempts: item.attempts + 1 });
|
||||
} else {
|
||||
totalAbandoned += 1;
|
||||
lastError = { eventId: item.event.eventId, type: item.event.type, attempt: item.attempts, error: "oa event publish attempts exhausted" };
|
||||
ctx().logger("error", "oa_event_publish_abandoned", lastError);
|
||||
}
|
||||
}
|
||||
}).finally(() => {
|
||||
activePublishes = Math.max(0, activePublishes - 1);
|
||||
pumpOaEventQueue();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function postOaEvent(event: OaEventEnvelope): void {
|
||||
const runtime = ctx();
|
||||
totalEnqueued += 1;
|
||||
lastEnqueuedAt = runtime.nowIso();
|
||||
if (queuedEvents.length >= maxQueuedEvents) {
|
||||
const dropped = queuedEvents.shift();
|
||||
totalDropped += 1;
|
||||
lastError = { eventId: dropped?.event.eventId ?? "", type: dropped?.event.type ?? "", error: "oa event publish queue overflow" };
|
||||
runtime.logger("error", "oa_event_publish_queue_overflow", { ...lastError, pending: queuedEvents.length, maxQueuedEvents });
|
||||
}
|
||||
queuedEvents.push({ event, attempts: 1 });
|
||||
pumpOaEventQueue();
|
||||
}
|
||||
|
||||
function normalizeCommandText(text: string): string {
|
||||
|
||||
Reference in New Issue
Block a user