mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(code-queue): add issue 3 gateway diagnostics

Browse Source
This commit is contained in:
Codex
2026-05-19 03:32:16 +00:00
parent c00c695197
commit 9d46ca2531
18 changed files with 1494 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src-rs/main.rs
+219 -6
View File
@@ -1,9 +1,9 @@
#![recursion_limit = "512"]
use chrono::{SecondsFormat, Utc};
use chrono::{DateTime, SecondsFormat, Utc};
use postgres::{Client, NoTls};
use serde_json::{json, Map, Value};
use std::collections::VecDeque;
use std::collections::{HashSet, VecDeque};
use std::env;
use std::error::Error;
use std::fs::{create_dir_all, OpenOptions};
@@ -19,6 +19,7 @@ const MAX_LIST_LIMIT: i64 = 200;
const MAX_OUTPUT_PAGE_LIMIT: usize = 500;
const MAX_PREVIEW_CHARS: usize = 6000;
const WORKDIR_MAX_CHARS: usize = 512;
const SCHEDULER_HEARTBEAT_STALE_MS: i64 = 5 * 60 * 1000;
#[derive(Clone)]
struct Config {
@@ -556,6 +557,10 @@ fn terminal_status(status: &str) -> bool {
matches!(status, "succeeded" | "failed" | "canceled")
}
fn active_status(status: &str) -> bool {
matches!(status, "running" | "judging")
}
fn queued_prompt_editable(task: &TaskMeta) -> bool {
task.status == "queued"
&& task.started_at.is_none()
@@ -636,6 +641,188 @@ fn output_max_seq(task: &TaskMeta) -> i64 {
task.last_output_seq.max(from_output)
}
fn task_scheduler_heartbeat(task: &TaskMeta) -> Option<Value> {
task.task_json
.as_ref()
.and_then(|value| value.get("schedulerHeartbeat"))
.filter(|value| value.is_object())
.cloned()
}
fn parse_iso_ms(value: Option<&str>) -> Option<i64> {
value
.and_then(|text| DateTime::parse_from_rfc3339(text).ok())
.map(|date| date.timestamp_millis())
}
fn heartbeat_string_field(heartbeat: &Value, key: &str) -> Option<String> {
heartbeat.get(key).and_then(Value::as_str).map(|value| value.to_string())
}
fn heartbeat_task_id(heartbeat: &Value) -> String {
heartbeat_string_field(heartbeat, "taskId").unwrap_or_default()
}
fn heartbeat_fresh(heartbeat: &Value, now_ms: i64) -> bool {
parse_iso_ms(heartbeat.get("lastLocalHeartbeatAt").and_then(Value::as_str))
.map(|at| now_ms.saturating_sub(at) < SCHEDULER_HEARTBEAT_STALE_MS)
.unwrap_or(false)
}
fn max_timestamp(values: Vec<Option<String>>) -> Value {
let mut best: Option<(String, i64)> = None;
for value in values.into_iter().flatten() {
let Some(ms) = parse_iso_ms(Some(&value)) else { continue };
if best.as_ref().map(|(_, best_ms)| ms >= *best_ms).unwrap_or(true) {
best = Some((value, ms));
}
}
best.map(|(value, _)| json!(value)).unwrap_or(Value::Null)
}
fn task_has_trace_gap(task: &TaskMeta, heartbeat: &Value, now_ms: i64) -> bool {
let trace_ms = parse_iso_ms(heartbeat.get("lastPersistedTraceAt").and_then(Value::as_str));
let output_seq = heartbeat.get("outputMaxSeq").and_then(Value::as_i64).unwrap_or_else(|| output_max_seq(task));
trace_ms
.map(|at| now_ms.saturating_sub(at) >= SCHEDULER_HEARTBEAT_STALE_MS && output_seq > 0)
.unwrap_or(false)
}
fn execution_diagnostics_from_tasks(tasks: &[TaskMeta], now: &str) -> Value {
let now_ms = parse_iso_ms(Some(now)).unwrap_or_else(|| Utc::now().timestamp_millis());
let mut database_active_task_ids: Vec<String> = tasks.iter().filter(|task| active_status(&task.status)).map(|task| task.id.clone()).collect();
database_active_task_ids.sort();
let mut active_heartbeats: Vec<(String, Value)> = tasks
.iter()
.filter_map(|task| task_scheduler_heartbeat(task).map(|heartbeat| (task.id.clone(), heartbeat)))
.collect();
active_heartbeats.sort_by(|left, right| left.0.cmp(&right.0));
let mut active_heartbeat_task_ids: Vec<String> = active_heartbeats.iter().map(|(_, heartbeat)| heartbeat_task_id(heartbeat)).filter(|id| !id.is_empty()).collect();
active_heartbeat_task_ids.sort();
active_heartbeat_task_ids.dedup();
let mut heartbeat_fresh_task_ids: Vec<String> = active_heartbeats
.iter()
.filter(|(_, heartbeat)| heartbeat_fresh(heartbeat, now_ms))
.map(|(_, heartbeat)| heartbeat_task_id(heartbeat))
.filter(|id| !id.is_empty())
.collect();
heartbeat_fresh_task_ids.sort();
heartbeat_fresh_task_ids.dedup();
let mut heartbeat_expired_task_ids: Vec<String> = active_heartbeats
.iter()
.filter(|(task_id, heartbeat)| {
tasks.iter().any(|task| &task.id == task_id && active_status(&task.status))
&& !heartbeat_fresh(heartbeat, now_ms)
})
.map(|(_, heartbeat)| heartbeat_task_id(heartbeat))
.filter(|id| !id.is_empty())
.collect();
heartbeat_expired_task_ids.sort();
heartbeat_expired_task_ids.dedup();
let heartbeat_task_set: HashSet<String> = active_heartbeat_task_ids.iter().cloned().collect();
let heartbeat_fresh_set: HashSet<String> = heartbeat_fresh_task_ids.iter().cloned().collect();
let heartbeat_expired_set: HashSet<String> = heartbeat_expired_task_ids.iter().cloned().collect();
let heartbeat_missing_task_ids: Vec<String> = database_active_task_ids
.iter()
.filter(|task_id| !heartbeat_task_set.contains(*task_id))
.cloned()
.collect();
let stale_recovery_candidate_task_ids: Vec<String> = database_active_task_ids
.iter()
.filter(|task_id| heartbeat_expired_set.contains(*task_id))
.cloned()
.collect();
let mut trace_gap_task_ids: Vec<String> = active_heartbeats
.iter()
.filter(|(task_id, heartbeat)| {
tasks.iter().find(|task| &task.id == task_id).map(|task| task_has_trace_gap(task, heartbeat, now_ms)).unwrap_or(false)
})
.map(|(_, heartbeat)| heartbeat_task_id(heartbeat))
.filter(|id| !id.is_empty())
.collect();
trace_gap_task_ids.sort();
trace_gap_task_ids.dedup();
let trace_gap_not_stale_task_ids: Vec<String> = trace_gap_task_ids
.iter()
.filter(|task_id| heartbeat_fresh_set.contains(*task_id))
.cloned()
.collect();
let split_brain = !database_active_task_ids.is_empty() && !heartbeat_fresh_task_ids.is_empty();
let stale_active = !stale_recovery_candidate_task_ids.is_empty();
let degraded = split_brain
|| stale_active
|| !heartbeat_missing_task_ids.is_empty()
|| !heartbeat_expired_task_ids.is_empty()
|| !trace_gap_task_ids.is_empty();
let state = if split_brain {
"split-brain"
} else if stale_active {
"stale-active"
} else if degraded {
"degraded"
} else {
"healthy"
};
let mut reasons = Vec::new();
if split_brain {
reasons.push("postgres control-plane has database-active tasks while its local active slots are empty, but scheduler heartbeat is fresh");
}
if stale_active {
reasons.push("owner heartbeat is expired and scheduler has no local active run for at least one database-active task");
}
if !heartbeat_missing_task_ids.is_empty() {
reasons.push("database-active tasks are missing scheduler-owned heartbeat");
}
if !trace_gap_not_stale_task_ids.is_empty() {
reasons.push("trace progress is stale while scheduler heartbeat is fresh; this is a trace gap, not stale active");
}
json!({
"state": state,
"health": state,
"degraded": degraded,
"splitBrain": split_brain,
"executionStateSource": "postgres-control-plane",
"controlPlane": "master-code-queue-mgr",
"databaseActiveTaskIds": database_active_task_ids,
"databaseActiveTaskCount": database_active_task_ids.len(),
"schedulerActiveTaskIds": [],
"schedulerActiveTaskCount": 0,
"schedulerActiveRunSlotCount": 0,
"schedulerActiveQueueIds": [],
"schedulerProcessingQueueIds": [],
"schedulerOrphanedActiveTaskIds": database_active_task_ids,
"schedulerOrphanedActiveTaskCount": database_active_task_ids.len(),
"activeHeartbeatTaskIds": active_heartbeat_task_ids,
"activeHeartbeatCount": active_heartbeat_task_ids.len(),
"heartbeatFreshTaskIds": heartbeat_fresh_task_ids,
"heartbeatExpiredTaskIds": heartbeat_expired_task_ids,
"heartbeatMissingTaskIds": heartbeat_missing_task_ids,
"staleRecoveryCandidateTaskIds": stale_recovery_candidate_task_ids,
"traceGapTaskIds": trace_gap_task_ids,
"traceGapNotStaleTaskIds": trace_gap_not_stale_task_ids,
"schedulerHeartbeatStaleMs": SCHEDULER_HEARTBEAT_STALE_MS,
"now": now,
"lastSchedulerHeartbeatAt": max_timestamp(active_heartbeats.iter().map(|(_, heartbeat)| heartbeat_string_field(heartbeat, "lastLocalHeartbeatAt")).collect()),
"lastObservedAgentEventAt": max_timestamp(active_heartbeats.iter().map(|(_, heartbeat)| heartbeat_string_field(heartbeat, "lastObservedAgentEventAt")).collect()),
"lastPersistedTraceAt": max_timestamp(active_heartbeats.iter().map(|(_, heartbeat)| heartbeat_string_field(heartbeat, "lastPersistedTraceAt")).collect()),
"oaPublisher": Value::Null,
"reasons": reasons,
"guidance": [
"Do not use master activeRunSlotCount=0 alone to declare a task stuck.",
"Check PostgreSQL task status, scheduler active runs/slots, scheduler heartbeat, and trace/OA publisher progress together.",
"Trace gap with a fresh scheduler heartbeat must not trigger stale retry.",
"Stale recovery is allowed only when the scheduler has no local active run and the owner heartbeat is expired."
]
})
}
fn last_assistant_message(task: &TaskMeta) -> Value {
let final_text = final_response(task);
if !final_text.trim().is_empty() {
@@ -724,6 +911,7 @@ fn task_list_response(_state: &AppState, task: &TaskMeta, lite: bool) -> Value {
"currentMode": task.current_mode,
"judgeFailCount": task.task_json.as_ref().and_then(|value| value.get("judgeFailCount")).and_then(Value::as_i64).unwrap_or(0),
"judgeFailRetryLimit": 3,
"schedulerHeartbeat": task_scheduler_heartbeat(task).unwrap_or(Value::Null),
"codexThreadId": task.codex_thread_id,
"activeTurnId": task.active_turn_id,
"lastError": task.last_error,
@@ -936,7 +1124,6 @@ fn queue_summary(state: &AppState) -> Result<Value, String> {
let mut counts = Map::new();
let mut total = 0_i64;
let mut unread_terminal_total = 0_i64;
let mut database_active_task_ids = Vec::new();
for row in rows {
let id = safe_queue_id(row.get::<_, String>("id").as_str());
let status_counts = json!({
@@ -956,9 +1143,6 @@ fn queue_summary(state: &AppState) -> Result<Value, String> {
let unread_terminal = row.get::<_, i64>("unread_terminal");
total += queue_total;
unread_terminal_total += unread_terminal;
if let Some(active) = row.get::<_, Option<String>>("active_task_id") {
database_active_task_ids.push(active);
}
queues.push(json!({
"id": id,
"name": row.get::<_, String>("name"),
@@ -972,6 +1156,30 @@ fn queue_summary(state: &AppState) -> Result<Value, String> {
"updatedAt": row.get::<_, Option<String>>("updated_at")
}));
}
let active_rows = client
.query(
"
SELECT id, queue_id, status, provider_id, execution_mode, model, cwd, prompt, base_prompt,
reference_task_ids, reference_injection, reasoning_effort, max_attempts, current_attempt,
current_mode, codex_thread_id, active_turn_id,
to_json(created_at)#>>'{}' AS created_at,
to_json(updated_at)#>>'{}' AS updated_at,
to_json(started_at)#>>'{}' AS started_at,
to_json(finished_at)#>>'{}' AS finished_at,
to_json(read_at)#>>'{}' AS read_at,
last_error, last_judge, output_count, event_count, attempt_count, last_output_seq,
task_json - 'output' - 'events' - 'promptHistory' AS task_json
FROM unidesk_code_queue_tasks
WHERE status IN ('running', 'judging')
ORDER BY updated_at DESC, id ASC
",
&[],
)
.map_err(|error| error.to_string())?;
let active_tasks: Vec<TaskMeta> = active_rows.iter().map(|row| row_to_task(row, true)).collect();
let mut database_active_task_ids: Vec<String> = active_tasks.iter().map(|task| task.id.clone()).collect();
database_active_task_ids.sort();
let execution_diagnostics = execution_diagnostics_from_tasks(&active_tasks, &now_iso());
if !queues.iter().any(|queue| queue.get("id").and_then(Value::as_str) == Some(DEFAULT_QUEUE_ID)) {
let now = now_iso();
queues.insert(0, json!({
@@ -1009,6 +1217,8 @@ fn queue_summary(state: &AppState) -> Result<Value, String> {
"databaseActiveTaskIds": database_active_task_ids,
"databaseActiveTaskCount": database_active_task_ids.len(),
"executionStateSource": "postgres-control-plane",
"executionDiagnostics": execution_diagnostics,
"schedulerHeartbeatStaleMs": SCHEDULER_HEARTBEAT_STALE_MS,
"orphanedActiveTaskIds": [],
"orphanedActiveTaskCount": 0,
"processing": false,
@@ -1793,6 +2003,7 @@ fn task_summary(task: &TaskMeta) -> Value {
"currentMode": task.current_mode,
"judgeFailCount": task.task_json.as_ref().and_then(|value| value.get("judgeFailCount")).and_then(Value::as_i64).unwrap_or(0),
"judgeFailRetryLimit": 3,
"schedulerHeartbeat": task_scheduler_heartbeat(task).unwrap_or(Value::Null),
"codexThreadId": task.codex_thread_id,
"activeTurnId": task.active_turn_id,
"createdAt": task.created_at,
@@ -1835,6 +2046,7 @@ fn trace_summary(task: &TaskMeta) -> Value {
"stepCount": task.task_json.as_ref().and_then(|value| value.get("stepCount")).and_then(Value::as_i64).unwrap_or(outputs.len() as i64),
"retainedStepCount": outputs.len(),
"outputMaxSeq": output_max_seq(task),
"schedulerHeartbeat": task_scheduler_heartbeat(task).unwrap_or(Value::Null),
"statsSource": "code-queue-mgr-rust-postgres",
"attempts": attempts_array(task),
"prompt": { "initialPreview": preview(&task.prompt, 1600), "basePreview": preview(&task.base_prompt, 1600), "chars": task.prompt.chars().count(), "lines": task.prompt.lines().count() },
@@ -2144,6 +2356,7 @@ fn health(state: &AppState) -> (Value, u16) {
"schemaReady": ready,
"schemaLastError": last_error,
"taskCount": task_count,
"queue": if ready { queue_summary(state).unwrap_or(Value::Null) } else { Value::Null },
"resourceBudget": {
"targetMemoryMb": 100,
"mgrPoolMax": state.config.mgr_pool_max,
src/components/microservices/code-queue-mgr/src/index.ts
+112
View File
@@ -166,6 +166,7 @@ interface QueueTask {
currentMode: RunMode | null;
codexThreadId: string | null;
activeTurnId: string | null;
schedulerHeartbeat?: JsonRecord | null;
finalResponse: string;
stepCount?: number;
llmStepCount?: number;
@@ -287,6 +288,7 @@ const workdirMaxLength = 512;
const defaultCodeModels = ["gpt-5.5", "gpt-5.4-mini", "gpt-5.4", "minimax-m2.7"];
const codeExecutionModes: CodeExecutionMode[] = ["default", "windows-native"];
const codexStatsTimeZone = "Asia/Shanghai";
const schedulerHeartbeatStaleMs = 5 * 60 * 1000;
const codexStatsDateFormatter = new Intl.DateTimeFormat("en-CA", {
timeZone: codexStatsTimeZone,
year: "numeric",
@@ -808,6 +810,108 @@ function outputMaxSeq(task: QueueTask): number {
return Math.max(0, numberField(task.outputMaxSeq, 0), ...outputSeq, ...promptSeq, ...attemptSeq);
}
function activeTask(task: QueueTask): boolean {
return task.status === "running" || task.status === "judging";
}
function schedulerHeartbeat(task: QueueTask): JsonRecord | null {
const heartbeat = asRecord(task.schedulerHeartbeat);
return heartbeat !== null && heartbeat.taskId === task.id ? heartbeat as JsonRecord : null;
}
function heartbeatTimestamp(heartbeat: JsonRecord | null, key: string): string | null {
const value = heartbeat?.[key];
return typeof value === "string" && timestampMs(value) !== null ? value : null;
}
function maxTimestamp(values: Array<string | null | undefined>): string | null {
let best: string | null = null;
let bestMs = -Infinity;
for (const value of values) {
const ms = timestampMs(value ?? null);
if (ms === null || ms < bestMs) continue;
best = value ?? null;
bestMs = ms;
}
return best;
}
function heartbeatFresh(heartbeat: JsonRecord | null, nowMs: number): boolean {
const atMs = timestampMs(heartbeatTimestamp(heartbeat, "lastLocalHeartbeatAt"));
return atMs !== null && nowMs - atMs < schedulerHeartbeatStaleMs;
}
function traceGap(task: QueueTask, heartbeat: JsonRecord | null, nowMs: number): boolean {
const traceAt = timestampMs(heartbeatTimestamp(heartbeat, "lastPersistedTraceAt"));
return traceAt !== null && nowMs - traceAt >= schedulerHeartbeatStaleMs && outputMaxSeq(task) > 0;
}
function controlPlaneExecutionDiagnostics(tasks: QueueTask[], now = nowIso()): JsonRecord {
const nowMs = timestampMs(now) ?? Date.now();
const databaseActiveTaskIds = tasks.filter(activeTask).map((task) => task.id).sort();
const heartbeatRows = tasks.map((task) => ({ task, heartbeat: schedulerHeartbeat(task) })).filter((row): row is { task: QueueTask; heartbeat: JsonRecord } => row.heartbeat !== null);
const activeHeartbeatTaskIds = heartbeatRows.map((row) => String(row.heartbeat.taskId ?? row.task.id)).sort();
const heartbeatFreshTaskIds = heartbeatRows.filter((row) => heartbeatFresh(row.heartbeat, nowMs)).map((row) => String(row.heartbeat.taskId ?? row.task.id)).sort();
const heartbeatFreshSet = new Set(heartbeatFreshTaskIds);
const heartbeatExpiredTaskIds = heartbeatRows.filter((row) => activeTask(row.task) && !heartbeatFreshSet.has(row.task.id)).map((row) => row.task.id).sort();
const activeHeartbeatSet = new Set(activeHeartbeatTaskIds);
const heartbeatMissingTaskIds = databaseActiveTaskIds.filter((taskId) => !activeHeartbeatSet.has(taskId));
const staleRecoveryCandidateTaskIds = heartbeatExpiredTaskIds.slice();
const traceGapTaskIds = heartbeatRows.filter((row) => traceGap(row.task, row.heartbeat, nowMs)).map((row) => row.task.id).sort();
const traceGapNotStaleTaskIds = traceGapTaskIds.filter((taskId) => heartbeatFreshSet.has(taskId));
const splitBrain = databaseActiveTaskIds.length > 0 && heartbeatFreshTaskIds.length > 0;
const staleActive = staleRecoveryCandidateTaskIds.length > 0;
const degraded = splitBrain
|| staleActive
|| heartbeatMissingTaskIds.length > 0
|| heartbeatExpiredTaskIds.length > 0
|| traceGapTaskIds.length > 0;
const state = splitBrain ? "split-brain" : staleActive ? "stale-active" : degraded ? "degraded" : "healthy";
const reasons: string[] = [];
if (splitBrain) reasons.push("postgres control-plane has database-active tasks while its local active slots are empty, but scheduler heartbeat is fresh");
if (staleActive) reasons.push("owner heartbeat is expired and scheduler has no local active run for at least one database-active task");
if (heartbeatMissingTaskIds.length > 0) reasons.push("database-active tasks are missing scheduler-owned heartbeat");
if (traceGapNotStaleTaskIds.length > 0) reasons.push("trace progress is stale while scheduler heartbeat is fresh; this is a trace gap, not stale active");
return {
state,
health: state,
degraded,
splitBrain,
executionStateSource: "postgres-control-plane",
controlPlane: "master-code-queue-mgr",
databaseActiveTaskIds,
databaseActiveTaskCount: databaseActiveTaskIds.length,
schedulerActiveTaskIds: [],
schedulerActiveTaskCount: 0,
schedulerActiveRunSlotCount: 0,
schedulerActiveQueueIds: [],
schedulerProcessingQueueIds: [],
schedulerOrphanedActiveTaskIds: databaseActiveTaskIds,
schedulerOrphanedActiveTaskCount: databaseActiveTaskIds.length,
activeHeartbeatTaskIds,
activeHeartbeatCount: activeHeartbeatTaskIds.length,
heartbeatFreshTaskIds,
heartbeatExpiredTaskIds,
heartbeatMissingTaskIds,
staleRecoveryCandidateTaskIds,
traceGapTaskIds,
traceGapNotStaleTaskIds,
schedulerHeartbeatStaleMs,
now,
lastSchedulerHeartbeatAt: maxTimestamp(heartbeatRows.map((row) => heartbeatTimestamp(row.heartbeat, "lastLocalHeartbeatAt"))),
lastObservedAgentEventAt: maxTimestamp(heartbeatRows.map((row) => heartbeatTimestamp(row.heartbeat, "lastObservedAgentEventAt"))),
lastPersistedTraceAt: maxTimestamp(heartbeatRows.map((row) => heartbeatTimestamp(row.heartbeat, "lastPersistedTraceAt"))),
oaPublisher: null,
reasons,
guidance: [
"Do not use master activeRunSlotCount=0 alone to declare a task stuck.",
"Check PostgreSQL task status, scheduler active runs/slots, scheduler heartbeat, and trace/OA publisher progress together.",
"Trace gap with a fresh scheduler heartbeat must not trigger stale retry.",
"Stale recovery is allowed only when the scheduler has no local active run and the owner heartbeat is expired.",
],
};
}
function normalizeOutput(value: unknown): LiveOutput[] {
if (!Array.isArray(value)) return [];
return value.flatMap((item): LiveOutput[] => {
@@ -876,6 +980,7 @@ function normalizeTask(value: unknown): QueueTask {
currentMode: record.currentMode === "initial" || record.currentMode === "retry" ? record.currentMode : null,
codexThreadId: typeof record.codexThreadId === "string" ? record.codexThreadId : null,
activeTurnId: typeof record.activeTurnId === "string" ? record.activeTurnId : null,
schedulerHeartbeat: asRecord(record.schedulerHeartbeat) as JsonRecord | null,
finalResponse: typeof record.finalResponse === "string" ? record.finalResponse : "",
lastError: typeof record.lastError === "string" ? record.lastError : null,
lastJudge: toJsonValue(record.lastJudge ?? null),
@@ -1153,6 +1258,7 @@ async function createTaskFromRequest(value: unknown): Promise<QueueTask> {
currentMode: null,
codexThreadId: null,
activeTurnId: null,
schedulerHeartbeat: null,
finalResponse: "",
stepCount: 0,
llmStepCount: 0,
@@ -1581,6 +1687,8 @@ async function queueSummary(tasks?: QueueTask[], queueRecords?: QueueRecord[]):
databaseActiveTaskIds,
databaseActiveTaskCount: databaseActiveTaskIds.length,
executionStateSource: "postgres-control-plane",
executionDiagnostics: controlPlaneExecutionDiagnostics(taskRows),
schedulerHeartbeatStaleMs,
orphanedActiveTaskIds: [],
orphanedActiveTaskCount: 0,
processing: false,
@@ -1817,6 +1925,7 @@ function taskListResponse(task: QueueTask, lite = true): JsonRecord {
currentMode: task.currentMode,
judgeFailCount: task.judgeFailCount,
judgeFailRetryLimit: 3,
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
codexThreadId: task.codexThreadId,
activeTurnId: task.activeTurnId,
lastError: task.lastError,
@@ -2280,6 +2389,7 @@ function traceSummary(task: QueueTask): JsonRecord {
stepCount: numberField(task.stepCount ?? task.llmStepCount, steps.length),
retainedStepCount: steps.length,
outputMaxSeq: outputMaxSeq(task),
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
statsSource: "code-queue-mgr-postgres",
attempts: task.attempts.map((attempt) => ({
index: attempt.index,
@@ -2766,6 +2876,7 @@ async function route(req: Request): Promise<Response> {
try {
if (url.pathname === "/" || url.pathname === "/health" || url.pathname === "/live") {
const taskCount = schemaReady ? Number((await traceSql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM unidesk_code_queue_tasks`)[0]?.count ?? 0) : null;
const queue = schemaReady ? await queueSummary() : null;
return jsonResponse({
ok: schemaReady,
service: "code-queue-mgr",
@@ -2780,6 +2891,7 @@ async function route(req: Request): Promise<Response> {
schemaReady,
schemaLastError,
taskCount,
queue,
resourceBudget: {
targetMemoryMb: 100,
mgrPoolMax: config.mgrDatabasePoolMax,
src/components/microservices/code-queue/src/code-agent/codex.ts
+13 -2
View File
@@ -20,6 +20,7 @@ export interface CodexPortContext {
queueIdOf: (task: QueueTask) => string;
recordNumberField: (record: Record<string, unknown> | null, keys: string[]) => number | null;
recordStringField: (record: Record<string, unknown> | null, keys: string[]) => string;
refreshSchedulerHeartbeat: (task: QueueTask, options?: { observedAgentEvent?: boolean; persistedTrace?: boolean }) => void;
remoteAppServerCommand: (task: QueueTask) => string;
windowsNativeAppServerCommand: (task: QueueTask) => string;
resolveReasoningEffort: (model: string, explicit?: string | null) => string | null;
@@ -322,16 +323,19 @@ function hasRecentCommandOutputDelta(task: QueueTask, itemId: string | undefined
function handleNotification(task: QueueTask, message: Record<string, unknown>, terminal: (status: TerminalStatus, error: string | null) => void): void {
const method = typeof message.method === "string" ? message.method : "unknown";
const params = extractRecord(message.params);
ctx().refreshSchedulerHeartbeat(task, { observedAgentEvent: true });
ctx().addEvent(task, eventSummary(message));
if (method === "thread/started") {
const threadId = extractString(extractRecord(params?.thread), "id");
if (threadId !== null) task.codexThreadId = threadId;
ctx().refreshSchedulerHeartbeat(task, { observedAgentEvent: true });
ctx().appendOutput(task, "system", `thread started ${threadId ?? "unknown"}\n`, method);
return;
}
if (method === "turn/started") {
const turnId = extractString(extractRecord(params?.turn), "id");
task.activeTurnId = turnId;
ctx().refreshSchedulerHeartbeat(task, { observedAgentEvent: true });
ctx().appendOutput(task, "system", `turn started ${turnId ?? "unknown"}\n`, method);
return;
}
@@ -383,6 +387,7 @@ function handleNotification(task: QueueTask, message: Record<string, unknown>, t
const status = terminalStatus(String(turn?.status ?? "failed"));
const error = extractRecord(turn?.error);
task.activeTurnId = null;
ctx().refreshSchedulerHeartbeat(task, { observedAgentEvent: true });
ctx().appendOutput(task, status === "completed" ? "system" : "error", `turn completed status=${status ?? "unknown"}\n`, method);
terminal(status, typeof error?.message === "string" ? error.message : null);
}
@@ -416,11 +421,17 @@ export async function runCodexTurn(task: QueueTask, prompt: string): Promise<Cod
await app.initialize();
const threadId = await app.startOrResumeThread();
task.codexThreadId = threadId;
ctx().activeRuns.set(queueId, { taskId: task.id, queueId, app, port: "codex", threadId, turnId: null });
const runStartedAt = ctx().nowIso();
ctx().activeRuns.set(queueId, { taskId: task.id, queueId, app, port: "codex", threadId, turnId: null, startedAt: runStartedAt, lastLocalHeartbeatAt: runStartedAt, lastObservedAgentEventAt: null, lastPersistedTraceAt: null });
ctx().refreshSchedulerHeartbeat(task);
const turnId = await app.startTurn(threadId, prompt);
task.activeTurnId = turnId;
const run = ctx().activeRuns.get(queueId);
if (run?.app === app) run.turnId = turnId;
if (run?.app === app) {
run.turnId = turnId;
run.lastLocalHeartbeatAt = ctx().nowIso();
}
ctx().refreshSchedulerHeartbeat(task);
ctx().persistTaskState(task);
const activityWatchdog = setInterval(() => {
if (terminalSeen) return;
src/components/microservices/code-queue/src/code-agent/common.ts
+4
View File
@@ -17,6 +17,10 @@ export interface ActiveRun {
port: CodeAgentPortKind;
threadId: string | null;
turnId: string | null;
startedAt: string;
lastLocalHeartbeatAt: string;
lastObservedAgentEventAt: string | null;
lastPersistedTraceAt: string | null;
}
export interface ActiveRunSlotWaiter {
src/components/microservices/code-queue/src/code-agent/opencode.ts
+11 -2
View File
@@ -25,6 +25,7 @@ export interface OpenCodePortContext {
providerIsMain: (providerId: string) => boolean;
queueIdOf: (task: QueueTask) => string;
recordStringField: (record: Record<string, unknown> | null, keys: string[], max?: number) => string;
refreshSchedulerHeartbeat: (task: QueueTask, options?: { observedAgentEvent?: boolean; persistedTrace?: boolean }) => void;
remoteHostWorkdirForTask: (task: QueueTask) => string;
safePreview: (value: string, max?: number) => string;
shellQuote: (value: string) => string;
@@ -238,6 +239,7 @@ class OpenCodeRunClient implements CodeAgentClient {
const trimmed = String(line).trim();
if (trimmed.length === 0) continue;
this.lastActivityAt = Date.now();
ctx().refreshSchedulerHeartbeat(this.task, { observedAgentEvent: true });
this.handleLine(trimmed);
}
} catch (error) {
@@ -295,7 +297,12 @@ class OpenCodeRunClient implements CodeAgentClient {
this.sessionId = sessionId;
if (this.task.codexThreadId === null) this.task.codexThreadId = sessionId;
const run = ctx().activeRuns.get(ctx().queueIdOf(this.task));
if (run?.app === this) run.threadId = sessionId;
if (run?.app === this) {
run.threadId = sessionId;
run.lastLocalHeartbeatAt = ctx().nowIso();
run.lastObservedAgentEventAt = run.lastLocalHeartbeatAt;
}
ctx().refreshSchedulerHeartbeat(this.task, { observedAgentEvent: true });
if (this.sessionAnnounced) return;
const resumed = this.task.currentMode === "retry" || this.task.attempts.length > 0;
ctx().appendOutput(this.task, "system", `opencode session ${resumed ? "resumed" : "started"} ${sessionId}\n`, resumed ? "opencode/session-resume" : "opencode/session-start");
@@ -363,8 +370,10 @@ async function runOpenCodeTurnOnce(task: QueueTask, prompt: string): Promise<Cod
}
await ctx().ensureTaskExecutionContainer(task);
const app = new OpenCodeRunClient(task, prompt);
ctx().activeRuns.set(queueId, { taskId: task.id, queueId, app, port: "opencode", threadId: task.codexThreadId, turnId: app.runId });
const runStartedAt = ctx().nowIso();
ctx().activeRuns.set(queueId, { taskId: task.id, queueId, app, port: "opencode", threadId: task.codexThreadId, turnId: app.runId, startedAt: runStartedAt, lastLocalHeartbeatAt: runStartedAt, lastObservedAgentEventAt: null, lastPersistedTraceAt: null });
task.activeTurnId = null;
ctx().refreshSchedulerHeartbeat(task);
ctx().persistTaskState(task);
const activityWatchdog = setInterval(() => {
const idleMs = Date.now() - app.lastActivityAt;
src/components/microservices/code-queue/src/execution-diagnostics.ts
+221
View File
@@ -0,0 +1,221 @@
// 重构前 index.ts 只读参考：commit 6a04144d3f5103014f75b637d7e6bc2f45bf007f，blob 56e590c1a6b5ca7ad128bf2c992f60e46c355a58；可用 `git show 6a04144d3f5103014f75b637d7e6bc2f45bf007f:src/components/microservices/code-queue/src/index.ts` 查看。
import type { ActiveRun } from "./code-agent/common";
import type { CodeQueueExecutionDiagnostics, JsonValue, QueueTask, SchedulerActiveRunHeartbeat } from "./types";
export const schedulerHeartbeatStaleMs = 5 * 60 * 1000;
interface ExecutionDiagnosticsInput {
now?: string;
heartbeatStaleMs?: number;
controlPlane: string;
executionStateSource: "scheduler-execution-plane" | "postgres-control-plane";
tasks: QueueTask[];
activeRuns?: Iterable<ActiveRun>;
activeRunSlotCount?: number;
activeQueueIds?: string[];
processingQueueIds?: string[];
orphanedActiveTaskIds?: string[];
oaPublisher?: JsonValue | null;
}
interface StaleRecoveryInput {
now?: string;
heartbeatStaleMs?: number;
task: QueueTask;
localActive: boolean;
}
function timestampMs(value: string | null | undefined): number | null {
if (typeof value !== "string" || value.length === 0) return null;
const time = Date.parse(value);
return Number.isFinite(time) ? time : null;
}
function maxTimestamp(values: Array<string | null | undefined>): string | null {
let best: string | null = null;
let bestMs = -Infinity;
for (const value of values) {
const ms = timestampMs(value);
if (ms === null || ms < bestMs) continue;
best = value ?? null;
bestMs = ms;
}
return best;
}
function activeTask(task: QueueTask): boolean {
return task.status === "running" || task.status === "judging";
}
function taskHeartbeat(task: QueueTask): SchedulerActiveRunHeartbeat | null {
const heartbeat = task.schedulerHeartbeat;
if (heartbeat === undefined || heartbeat === null || heartbeat.taskId !== task.id) return null;
return heartbeat;
}
function heartbeatFresh(heartbeat: SchedulerActiveRunHeartbeat | null, nowMs: number, staleMs: number): boolean {
const atMs = timestampMs(heartbeat?.lastLocalHeartbeatAt);
return atMs !== null && nowMs - atMs < staleMs;
}
function outputMaxSeq(task: QueueTask): number {
const values = [
Number(task.outputMaxSeq ?? 0),
...((Array.isArray(task.output) ? task.output : []).map((item) => Number(item.seq ?? 0))),
...((Array.isArray(task.attempts) ? task.attempts : []).map((item) => Number(item.outputEndSeq ?? 0))),
].filter((value) => Number.isFinite(value) && value >= 0);
return values.length > 0 ? Math.max(...values.map((value) => Math.floor(value))) : 0;
}
export function staleRecoveryCandidate(input: StaleRecoveryInput): { allowed: boolean; reason: string; heartbeatFresh: boolean; heartbeatAgeMs: number | null } {
if (!activeTask(input.task)) return { allowed: false, reason: "not-db-active", heartbeatFresh: false, heartbeatAgeMs: null };
if (input.localActive) return { allowed: false, reason: "scheduler-local-active", heartbeatFresh: true, heartbeatAgeMs: 0 };
const nowMs = timestampMs(input.now) ?? Date.now();
const heartbeat = taskHeartbeat(input.task);
if (heartbeat === null) return { allowed: false, reason: "owner-heartbeat-missing", heartbeatFresh: false, heartbeatAgeMs: null };
const heartbeatMs = timestampMs(heartbeat.lastLocalHeartbeatAt);
if (heartbeatMs === null) return { allowed: false, reason: "owner-heartbeat-missing", heartbeatFresh: false, heartbeatAgeMs: null };
const heartbeatAgeMs = Math.max(0, nowMs - heartbeatMs);
const fresh = heartbeatAgeMs < (input.heartbeatStaleMs ?? schedulerHeartbeatStaleMs);
if (fresh) return { allowed: false, reason: "owner-heartbeat-fresh", heartbeatFresh: true, heartbeatAgeMs };
return { allowed: true, reason: "owner-heartbeat-expired", heartbeatFresh: false, heartbeatAgeMs };
}
export function taskHasTraceGapButFreshHeartbeat(task: QueueTask, now: string, heartbeatStaleMs = schedulerHeartbeatStaleMs): boolean {
if (!activeTask(task)) return false;
const heartbeat = taskHeartbeat(task);
if (!heartbeatFresh(heartbeat, timestampMs(now) ?? Date.now(), heartbeatStaleMs)) return false;
const traceAt = timestampMs(heartbeat?.lastPersistedTraceAt);
if (traceAt === null) return false;
return (timestampMs(now) ?? Date.now()) - traceAt >= heartbeatStaleMs;
}
export function buildSchedulerHeartbeat(task: QueueTask, run: ActiveRun | null, options: {
now: string;
owner: string;
schedulerInstance: string;
agentPort: string;
lastObservedAgentEventAt?: string | null;
lastPersistedTraceAt?: string | null;
}): SchedulerActiveRunHeartbeat {
const previous = taskHeartbeat(task);
return {
taskId: task.id,
queueId: task.queueId,
attempt: Number.isInteger(task.currentAttempt) && task.currentAttempt > 0 ? task.currentAttempt : 0,
activeTurnId: run?.turnId ?? task.activeTurnId ?? null,
codexThreadId: run?.threadId ?? task.codexThreadId ?? null,
owner: options.owner,
schedulerInstance: options.schedulerInstance,
executionPlane: "scheduler-execution-plane",
agentPort: run?.port ?? options.agentPort,
status: task.status,
lastLocalHeartbeatAt: options.now,
lastObservedAgentEventAt: options.lastObservedAgentEventAt ?? previous?.lastObservedAgentEventAt ?? null,
lastPersistedTraceAt: options.lastPersistedTraceAt ?? previous?.lastPersistedTraceAt ?? null,
outputMaxSeq: outputMaxSeq(task),
source: "scheduler",
};
}
export function buildExecutionDiagnostics(input: ExecutionDiagnosticsInput): CodeQueueExecutionDiagnostics {
const now = input.now ?? new Date().toISOString();
const nowMs = timestampMs(now) ?? Date.now();
const staleMs = input.heartbeatStaleMs ?? schedulerHeartbeatStaleMs;
const databaseActiveTaskIds = input.tasks.filter(activeTask).map((task) => task.id).sort();
const activeRunRows = Array.from(input.activeRuns ?? []);
const schedulerActiveTaskIds = Array.from(new Set(activeRunRows.map((run) => run.taskId).filter(Boolean))).sort();
const schedulerActiveTaskSet = new Set(schedulerActiveTaskIds);
const activeHeartbeats = input.tasks
.map((task) => taskHeartbeat(task))
.filter((heartbeat): heartbeat is SchedulerActiveRunHeartbeat => heartbeat !== null);
const heartbeatFreshTaskIds = activeHeartbeats
.filter((heartbeat) => heartbeatFresh(heartbeat, nowMs, staleMs))
.map((heartbeat) => heartbeat.taskId)
.sort();
const heartbeatFreshSet = new Set(heartbeatFreshTaskIds);
const heartbeatExpiredTaskIds = activeHeartbeats
.filter((heartbeat) => {
const heartbeatTask = input.tasks.find((task) => task.id === heartbeat.taskId);
return heartbeatTask !== undefined && activeTask(heartbeatTask) && !heartbeatFreshSet.has(heartbeat.taskId);
})
.map((heartbeat) => heartbeat.taskId)
.sort();
const activeHeartbeatTaskIds = activeHeartbeats.map((heartbeat) => heartbeat.taskId).sort();
const activeHeartbeatSet = new Set(activeHeartbeatTaskIds);
const heartbeatMissingTaskIds = databaseActiveTaskIds.filter((taskId) => !activeHeartbeatSet.has(taskId));
const staleRecoveryCandidateTaskIds = input.tasks
.filter((task) => staleRecoveryCandidate({ task, now, heartbeatStaleMs: staleMs, localActive: schedulerActiveTaskSet.has(task.id) }).allowed)
.map((task) => task.id)
.sort();
const traceGapTaskIds = activeHeartbeats
.filter((heartbeat) => {
const traceAt = timestampMs(heartbeat.lastPersistedTraceAt);
return traceAt !== null && nowMs - traceAt >= staleMs;
})
.map((heartbeat) => heartbeat.taskId)
.sort();
const traceGapNotStaleTaskIds = traceGapTaskIds.filter((taskId) => heartbeatFreshSet.has(taskId));
const schedulerLocalMissingIds = databaseActiveTaskIds.filter((taskId) => !schedulerActiveTaskSet.has(taskId));
const splitBrain = input.executionStateSource === "postgres-control-plane"
&& databaseActiveTaskIds.length > 0
&& schedulerActiveTaskIds.length === 0
&& heartbeatFreshTaskIds.length > 0;
const staleActive = staleRecoveryCandidateTaskIds.length > 0;
const degraded = splitBrain
|| staleActive
|| heartbeatMissingTaskIds.length > 0
|| heartbeatExpiredTaskIds.length > 0
|| schedulerLocalMissingIds.length > 0
|| traceGapTaskIds.length > 0
|| (typeof input.oaPublisher === "object" && input.oaPublisher !== null && !Array.isArray(input.oaPublisher) && (Number((input.oaPublisher as Record<string, JsonValue>).pending ?? 0) > 0 || (input.oaPublisher as Record<string, JsonValue>).lastError !== null));
const state = splitBrain ? "split-brain" : staleActive ? "stale-active" : degraded ? "degraded" : "healthy";
const lastSchedulerHeartbeatAt = maxTimestamp(activeHeartbeats.map((heartbeat) => heartbeat.lastLocalHeartbeatAt));
const lastObservedAgentEventAt = maxTimestamp(activeHeartbeats.map((heartbeat) => heartbeat.lastObservedAgentEventAt));
const lastPersistedTraceAt = maxTimestamp(activeHeartbeats.map((heartbeat) => heartbeat.lastPersistedTraceAt));
const reasons: string[] = [];
if (splitBrain) reasons.push("postgres control-plane has database-active tasks while its local active slots are empty, but scheduler heartbeat is fresh");
if (staleActive) reasons.push("owner heartbeat is expired and scheduler has no local active run for at least one database-active task");
if (heartbeatMissingTaskIds.length > 0) reasons.push("database-active tasks are missing scheduler-owned heartbeat");
if (traceGapNotStaleTaskIds.length > 0) reasons.push("trace progress is stale while scheduler heartbeat is fresh; this is a trace gap, not stale active");
const guidance = [
"Do not use master activeRunSlotCount=0 alone to declare a task stuck.",
"Check PostgreSQL task status, scheduler active runs/slots, scheduler heartbeat, and trace/OA publisher progress together.",
"Trace gap with a fresh scheduler heartbeat must not trigger stale retry.",
"Stale recovery is allowed only when the scheduler has no local active run and the owner heartbeat is expired.",
];
return {
state,
health: state,
degraded,
splitBrain,
executionStateSource: input.executionStateSource,
controlPlane: input.controlPlane,
databaseActiveTaskIds,
databaseActiveTaskCount: databaseActiveTaskIds.length,
schedulerActiveTaskIds,
schedulerActiveTaskCount: schedulerActiveTaskIds.length,
schedulerActiveRunSlotCount: input.activeRunSlotCount ?? schedulerActiveTaskIds.length,
schedulerActiveQueueIds: (input.activeQueueIds ?? activeRunRows.map((run) => run.queueId)).slice().sort(),
schedulerProcessingQueueIds: (input.processingQueueIds ?? []).slice().sort(),
schedulerOrphanedActiveTaskIds: (input.orphanedActiveTaskIds ?? schedulerLocalMissingIds).slice().sort(),
schedulerOrphanedActiveTaskCount: (input.orphanedActiveTaskIds ?? schedulerLocalMissingIds).length,
activeHeartbeatTaskIds,
activeHeartbeatCount: activeHeartbeatTaskIds.length,
heartbeatFreshTaskIds,
heartbeatExpiredTaskIds,
heartbeatMissingTaskIds,
staleRecoveryCandidateTaskIds,
traceGapTaskIds,
traceGapNotStaleTaskIds,
schedulerHeartbeatStaleMs: staleMs,
now,
lastSchedulerHeartbeatAt,
lastObservedAgentEventAt,
lastPersistedTraceAt,
oaPublisher: input.oaPublisher ?? null,
reasons,
guidance,
};
}
src/components/microservices/code-queue/src/index.ts
+148 -9
View File
@@ -25,6 +25,7 @@ import type {
RunMode,
RuntimeConfig,
CodeQueueServiceRole,
SchedulerActiveRunHeartbeat,
TaskStatus,
WorkdirRecord,
} from "./types";
@@ -114,6 +115,7 @@ import {
taskForListResponse,
transcriptChunkResponse,
} from "./queue-api";
import { buildExecutionDiagnostics, buildSchedulerHeartbeat, schedulerHeartbeatStaleMs, staleRecoveryCandidate } from "./execution-diagnostics";
import { ReferenceTaskLookupError, configureReferences, injectReferencedTaskContext, taskReferenceIds } from "./references";
import {
applyOaTraceStatsToTaskJson,
@@ -207,6 +209,14 @@ let shutdownRequested = false;
let serviceReady = false;
let sql: SqlClient = createSqlClient();
let sqlRotationInFlight: Promise<void> | null = null;
const schedulerReconcileAudit = {
lastRunAt: null as string | null,
lastReason: null as string | null,
lastLoaded: 0,
lastChanged: 0,
lastRecovered: 0,
lastError: null as string | null,
};
function createSqlClient(): SqlClient {
return postgres(config.databaseUrl, {
@@ -966,6 +976,30 @@ function taskCanHaveActiveTurn(status: TaskStatus): boolean {
return status === "running" || status === "judging";
}
function normalizeSchedulerHeartbeat(task: QueueTask): SchedulerActiveRunHeartbeat | null {
const heartbeat = task.schedulerHeartbeat;
if (heartbeat === null || heartbeat === undefined || typeof heartbeat !== "object") return null;
if (heartbeat.taskId !== task.id) return null;
if (typeof heartbeat.lastLocalHeartbeatAt !== "string" || timestampMs(heartbeat.lastLocalHeartbeatAt) === null) return null;
return {
taskId: task.id,
queueId: safeQueueId(heartbeat.queueId),
attempt: Number.isInteger(heartbeat.attempt) && heartbeat.attempt >= 0 ? heartbeat.attempt : Math.max(0, Number(task.currentAttempt || 0)),
activeTurnId: typeof heartbeat.activeTurnId === "string" && heartbeat.activeTurnId.length > 0 ? heartbeat.activeTurnId : null,
codexThreadId: typeof heartbeat.codexThreadId === "string" && heartbeat.codexThreadId.length > 0 ? heartbeat.codexThreadId : null,
owner: typeof heartbeat.owner === "string" && heartbeat.owner.length > 0 ? heartbeat.owner : config.instanceId,
schedulerInstance: typeof heartbeat.schedulerInstance === "string" && heartbeat.schedulerInstance.length > 0 ? heartbeat.schedulerInstance : config.instanceId,
executionPlane: "scheduler-execution-plane",
agentPort: typeof heartbeat.agentPort === "string" && heartbeat.agentPort.length > 0 ? heartbeat.agentPort : codeAgentPortForModel(task.model),
status: taskCanHaveActiveTurn(heartbeat.status) ? heartbeat.status : task.status,
lastLocalHeartbeatAt: heartbeat.lastLocalHeartbeatAt,
lastObservedAgentEventAt: typeof heartbeat.lastObservedAgentEventAt === "string" && timestampMs(heartbeat.lastObservedAgentEventAt) !== null ? heartbeat.lastObservedAgentEventAt : null,
lastPersistedTraceAt: typeof heartbeat.lastPersistedTraceAt === "string" && timestampMs(heartbeat.lastPersistedTraceAt) !== null ? heartbeat.lastPersistedTraceAt : null,
outputMaxSeq: Number.isInteger(heartbeat.outputMaxSeq) && heartbeat.outputMaxSeq >= 0 ? heartbeat.outputMaxSeq : taskViewOutputMaxSeq(task),
source: "scheduler",
};
}
function normalizeTask(task: QueueTask): QueueTask {
task.queueId = safeQueueId(task.queueId);
task.queueEnteredAt = taskQueueEnteredAt(task);
@@ -996,6 +1030,8 @@ function normalizeTask(task: QueueTask): QueueTask {
task.stepCount = stepCount;
task.llmStepCount = stepCount;
task.outputMaxSeq = Math.max(normalizeTaskMetric(task.outputMaxSeq) ?? 0, taskViewOutputMaxSeq(task));
task.schedulerHeartbeat = normalizeSchedulerHeartbeat(task);
if (!taskCanHaveActiveTurn(task.status)) task.schedulerHeartbeat = null;
pruneTaskHotState(task);
return task;
}
@@ -1044,6 +1080,29 @@ function publishQueueEvent(reason: string, queueId = ""): void {
publishCodeQueueQueueUpdated(queueId, reason);
}
function refreshSchedulerHeartbeat(task: QueueTask, options: { observedAgentEvent?: boolean; persistedTrace?: boolean } = {}): void {
if (!taskCanHaveActiveTurn(task.status)) {
task.schedulerHeartbeat = null;
return;
}
const run = activeRunForTask(task);
const now = nowIso();
if (run !== null) {
run.lastLocalHeartbeatAt = now;
if (options.observedAgentEvent === true) run.lastObservedAgentEventAt = now;
if (options.persistedTrace === true) run.lastPersistedTraceAt = now;
}
const previous = normalizeSchedulerHeartbeat(task);
task.schedulerHeartbeat = buildSchedulerHeartbeat(task, run, {
now,
owner: config.instanceId,
schedulerInstance: config.instanceId,
agentPort: codeAgentPortForModel(task.model),
lastObservedAgentEventAt: options.observedAgentEvent === true ? now : run?.lastObservedAgentEventAt ?? previous?.lastObservedAgentEventAt ?? null,
lastPersistedTraceAt: options.persistedTrace === true ? now : run?.lastPersistedTraceAt ?? previous?.lastPersistedTraceAt ?? null,
});
}
function isOpenCodeStepBoundaryMethod(method: string | undefined): boolean {
return method === "opencode/step-start" || method === "opencode/step-finish";
}
@@ -1228,6 +1287,8 @@ function markTaskDirty(taskId: string): void {
}
function persistTaskState(task: QueueTask): void {
if (taskCanHaveActiveTurn(task.status)) refreshSchedulerHeartbeat(task);
else task.schedulerHeartbeat = null;
markTaskDirty(task.id);
persistState(false);
publishTaskOaEvent(task, "persist");
@@ -2567,6 +2628,7 @@ function appendPromptHistory(task: QueueTask, output: LiveOutput | null, method:
function addEvent(task: QueueTask, event: CodexEventSummary): void {
task.events.push(event);
if (config.maxInMemoryEventRecords > 0 && task.events.length > config.maxInMemoryEventRecords) task.events.splice(0, task.events.length - config.maxInMemoryEventRecords);
refreshSchedulerHeartbeat(task, { observedAgentEvent: true });
markTaskDirty(task.id);
publishTaskOaEvent(task, "agent-event", { onlyStepChange: true });
}
@@ -2757,6 +2819,7 @@ configureTaskOutput({
markTaskDirty,
nowIso,
onOutputAppended: (task, output, op) => {
refreshSchedulerHeartbeat(task, { persistedTrace: true });
const archiveOp = op === "append" ? "append" : "set";
const stepChanged = recordTaskOutputMetrics(task, output, archiveOp);
const projectionOutput = traceStepOutputForProjection(task, output);
@@ -2832,12 +2895,14 @@ configureQueueApi({
parseTextLimit,
processing: () => processing,
processingQueues,
oaEventPublisherStatus: () => oaEventPublisherStatus() as JsonValue,
queueHeadTask,
queueIdOf,
queues: () => state.queues,
queueTaskIsRunnable,
queuedStatusReason,
queuedTaskPromptEditable,
reconcileStatus: schedulerReconcileStatus,
runGarbageCollection,
safeQueueId,
safeQueueName,
@@ -2908,6 +2973,7 @@ configureCodexPort({
queueIdOf,
recordNumberField,
recordStringField,
refreshSchedulerHeartbeat,
remoteAppServerCommand,
windowsNativeAppServerCommand,
resolveReasoningEffort,
@@ -2932,6 +2998,7 @@ configureOpenCodePort({
providerIsMain,
queueIdOf,
recordStringField,
refreshSchedulerHeartbeat,
remoteHostWorkdirForTask,
safePreview,
shellQuote,
@@ -3433,7 +3500,41 @@ function orphanedActiveTaskAgeMs(task: QueueTask): number {
function taskIsRecoverableOrphanedActive(task: QueueTask, staleMs = orphanedActiveTaskRecoveryStaleMs): boolean {
if (task.status !== "running" && task.status !== "judging") return false;
if (activeTaskHasLocalRunSlotOrWaiter(task)) return false;
return orphanedActiveTaskAgeMs(task) >= staleMs;
const decision = staleRecoveryCandidate({ task, localActive: false, heartbeatStaleMs: staleMs, now: nowIso() });
if (decision.allowed) return true;
if (decision.reason === "owner-heartbeat-missing") return orphanedActiveTaskAgeMs(task) >= staleMs;
return false;
}
function schedulerReconcileStatus(tasks: QueueTask[] = state.tasks): JsonValue {
const orphanedActiveTasks = tasks
.filter((task) => (task.status === "running" || task.status === "judging") && !activeTaskHasLocalRunSlotOrWaiter(task))
.sort((left, right) => left.id.localeCompare(right.id));
const recoverableOrphanedActiveTasks = orphanedActiveTasks
.filter((task) => taskIsRecoverableOrphanedActive(task))
.sort((left, right) => left.id.localeCompare(right.id));
const retryWaitTaskIds = tasks
.filter((task) => task.status === "retry_wait")
.map((task) => task.id)
.sort();
return {
ok: recoverableOrphanedActiveTasks.length === 0,
schedulerEnabled: config.schedulerEnabled,
pollIntervalMs: config.schedulerPollIntervalMs,
staleActiveRecoveryStaleMs: orphanedActiveTaskRecoveryStaleMs,
orphanedActiveTaskCount: orphanedActiveTasks.length,
orphanedActiveTaskIds: orphanedActiveTasks.map((task) => task.id),
recoverableOrphanedActiveTaskCount: recoverableOrphanedActiveTasks.length,
recoverableOrphanedActiveTaskIds: recoverableOrphanedActiveTasks.map((task) => task.id),
retryWaitTaskCount: retryWaitTaskIds.length,
retryWaitTaskIds,
lastRunAt: schedulerReconcileAudit.lastRunAt,
lastReason: schedulerReconcileAudit.lastReason,
lastLoaded: schedulerReconcileAudit.lastLoaded,
lastChanged: schedulerReconcileAudit.lastChanged,
lastRecovered: schedulerReconcileAudit.lastRecovered,
lastError: schedulerReconcileAudit.lastError,
};
}
function clearInactiveActiveTurnIds(): number {
@@ -3485,6 +3586,8 @@ function queueActiveTasksForRestartRetry(reason: string, method: string, options
}
async function recoverOrphanedActiveTasks(reason: string, method: string, options: { taskIds?: Set<string>; staleMs?: number } = {}): Promise<number> {
schedulerReconcileAudit.lastRunAt = nowIso();
schedulerReconcileAudit.lastReason = reason;
const staleMs = options.staleMs ?? orphanedActiveTaskRecoveryStaleMs;
const staleOrphanTaskIds = new Set(
state.tasks
@@ -3492,8 +3595,13 @@ async function recoverOrphanedActiveTasks(reason: string, method: string, option
.filter((task) => taskIsRecoverableOrphanedActive(task, staleMs))
.map((task) => task.id),
);
if (staleOrphanTaskIds.size === 0) return 0;
if (staleOrphanTaskIds.size === 0) {
schedulerReconcileAudit.lastRecovered = 0;
schedulerReconcileAudit.lastError = null;
return 0;
}
const recovered = queueActiveTasksForRestartRetry(reason, method, { taskIds: staleOrphanTaskIds });
schedulerReconcileAudit.lastRecovered = recovered;
if (recovered === 0) return 0;
logger("warn", "scheduler_orphaned_active_tasks_requeued", {
reason,
@@ -3502,6 +3610,7 @@ async function recoverOrphanedActiveTasks(reason: string, method: string, option
dirtyTaskCount: dirtyDatabaseTaskIds.size,
});
await flushDirtyTasksToDatabase(true);
schedulerReconcileAudit.lastError = null;
logger("info", "scheduler_orphaned_active_tasks_flushed", { reason, recovered });
scheduleQueue();
return recovered;
@@ -3567,6 +3676,7 @@ async function runTask(task: QueueTask): Promise<void> {
task.readAt = null;
task.finishedAt = null;
task.updatedAt = startedAt;
refreshSchedulerHeartbeat(task);
let claimed = false;
try {
claimed = await claimTaskInDatabase(task, claimQueueId);
@@ -4053,14 +4163,25 @@ function mergeSchedulerDatabaseTasks(tasks: QueueTask[]): number {
async function refreshSchedulerTasksFromDatabase(reason: string): Promise<number> {
if (!databaseReady || !config.schedulerEnabled) return 0;
const tasks = await loadTasksFromDatabase("hot");
const changed = mergeSchedulerDatabaseTasks(tasks);
const recovered = await recoverOrphanedActiveTasks("Scheduler recovered database active task without local run", `scheduler/${reason}-recovery`);
if (changed > 0) {
logger("info", "scheduler_database_hot_tasks_refreshed", { reason, changed, loaded: tasks.length });
scheduleQueue();
schedulerReconcileAudit.lastRunAt = nowIso();
schedulerReconcileAudit.lastReason = reason;
try {
const tasks = await loadTasksFromDatabase("hot");
const changed = mergeSchedulerDatabaseTasks(tasks);
const recovered = await recoverOrphanedActiveTasks("Scheduler recovered database active task without local run", `scheduler/${reason}-recovery`);
schedulerReconcileAudit.lastLoaded = tasks.length;
schedulerReconcileAudit.lastChanged = changed;
schedulerReconcileAudit.lastRecovered = recovered;
schedulerReconcileAudit.lastError = null;
if (changed > 0) {
logger("info", "scheduler_database_hot_tasks_refreshed", { reason, changed, loaded: tasks.length });
scheduleQueue();
}
return changed + recovered;
} catch (error) {
schedulerReconcileAudit.lastError = databaseErrorMessage(error);
throw error;
}
return changed + recovered;
}
function startSchedulerDatabasePoller(): void {
@@ -4081,6 +4202,23 @@ function activeRunForTask(task: QueueTask): ActiveRun | null {
return Array.from(activeRuns.values()).find((run) => run.taskId === task.id) ?? null;
}
function executionDiagnosticsForTasks(tasks: QueueTask[] = state.tasks): JsonValue {
return buildExecutionDiagnostics({
controlPlane: "D601-code-queue-scheduler",
executionStateSource: "scheduler-execution-plane",
tasks,
activeRuns: activeRuns.values(),
activeRunSlotCount: activeRunSlotCount(),
activeQueueIds: activeRunSlotQueueIds(),
processingQueueIds: Array.from(processingQueues),
orphanedActiveTaskIds: tasks
.filter((task) => (task.status === "running" || task.status === "judging") && activeRunForTask(task) === null)
.map((task) => task.id),
oaPublisher: oaEventPublisherStatus() as JsonValue,
heartbeatStaleMs: schedulerHeartbeatStaleMs,
}) as unknown as JsonValue;
}
function installShutdownHandlers(): void {
const stop = (signal: NodeJS.Signals): void => {
if (shutdownRequested) process.exit(0);
@@ -5053,6 +5191,7 @@ async function route(req: Request): Promise<Response> {
schedulerEnabled: config.schedulerEnabled,
schedulerPollIntervalMs: config.schedulerPollIntervalMs,
queue: queueSummary(false, state.tasks),
executionDiagnostics: executionDiagnosticsForTasks(state.tasks),
egressProxy: await providerGatewayEgressProxyStatus(),
oaEventPublisher: oaEventPublisherStatus(),
startedAt: serviceStartedAt,
src/components/microservices/code-queue/src/queue-api.ts
+21 -1
View File
@@ -6,13 +6,14 @@ import { claudeQqNotificationOutboxStats, notificationTargetConfigured, notifica
import { executionModeOptions, executionProviderOptions } from "./provider-runtime";
import { taskFullOutput } from "./task-output";
import { applyOaTraceStatsToTaskJson, taskScopeId, type OaTraceStats } from "./oa-events";
import { buildExecutionDiagnostics, schedulerHeartbeatStaleMs } from "./execution-diagnostics";
import { buildCompactTaskTranscript, buildTaskTranscript, cachedPreviewTranscript, fullTranscript, prefixPreview, safePreview, statsDaysFromUrl, taskForCompactMetaResponse, taskForMetaResponse, taskStatisticsSummary, taskTiming, timestampMs } from "./task-view";
import { userPromptForDisplay } from "./prompts";
import type { ActiveRun, ActiveRunSlotWaiter } from "./code-agent/common";
import type { JsonValue, QueueRecord, QueuedStatusReason, QueueTask, RuntimeConfig, TaskStatus, TranscriptLine } from "./types";
export interface QueueApiContext {
config: Pick<RuntimeConfig, "codeModels" | "codexModels" | "codexSqliteLogExportBatchSize" | "codexSqliteLogExportEnabled" | "codexSqliteLogExportIntervalMs" | "codexSqliteLogMaxBytes" | "defaultModel" | "defaultReasoningEffort" | "defaultWorkdir" | "judgeMaxTokens" | "judgeRepairAttempts" | "mainProviderId" | "maxActiveQueues" | "maxInMemoryEventRecords" | "maxInMemoryOutputRecords" | "minimaxApiKey" | "minimaxModel" | "modelReasoningEfforts" | "notifyClaudeQqBaseUrl" | "notifyClaudeQqEnabled" | "notifyClaudeQqMaxResponseChars" | "notifyClaudeQqRetryIntervalMs" | "notifyClaudeQqSendAttempts" | "notifyClaudeQqTargetType" | "notifyClaudeQqTimeoutMs" | "outputArchiveDir" | "remoteDefaultWorkdir" | "windowsNativeCodexDefaultWorkdir">;
config: Pick<RuntimeConfig, "codeModels" | "codexModels" | "codexSqliteLogExportBatchSize" | "codexSqliteLogExportEnabled" | "codexSqliteLogExportIntervalMs" | "codexSqliteLogMaxBytes" | "defaultModel" | "defaultReasoningEffort" | "defaultWorkdir" | "judgeMaxTokens" | "judgeRepairAttempts" | "mainProviderId" | "maxActiveQueues" | "maxInMemoryEventRecords" | "maxInMemoryOutputRecords" | "minimaxApiKey" | "minimaxModel" | "modelReasoningEfforts" | "notifyClaudeQqBaseUrl" | "notifyClaudeQqEnabled" | "notifyClaudeQqMaxResponseChars" | "notifyClaudeQqRetryIntervalMs" | "notifyClaudeQqSendAttempts" | "notifyClaudeQqTargetType" | "notifyClaudeQqTimeoutMs" | "outputArchiveDir" | "remoteDefaultWorkdir" | "schedulerEnabled" | "schedulerPollIntervalMs" | "windowsNativeCodexDefaultWorkdir">;
activeRunSlotQueueIds: () => string[];
activeRunSlotWaiterSummaries: () => JsonValue[];
activeRuns: Map<string, ActiveRun>;
@@ -34,12 +35,14 @@ export interface QueueApiContext {
parseTextLimit: (url: URL) => number;
processing: () => boolean;
processingQueues: Set<string>;
oaEventPublisherStatus: () => JsonValue;
queueHeadTask: (queueId: string, tasks?: QueueTask[]) => QueueTask | null;
queueIdOf: (task: QueueTask) => string;
queues: () => QueueRecord[];
queueTaskIsRunnable: (task: QueueTask) => boolean;
queuedStatusReason: (task: QueueTask, tasks?: QueueTask[]) => QueuedStatusReason | null;
queuedTaskPromptEditable: (task: QueueTask) => boolean;
reconcileStatus: (tasks: QueueTask[]) => JsonValue;
runGarbageCollection: () => void;
safeQueueId: (value: unknown) => string;
safeQueueName: (value: unknown, queueId: string) => string;
@@ -258,6 +261,7 @@ function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTa
currentMode: task.currentMode,
judgeFailCount: task.judgeFailCount,
judgeFailRetryLimit: ctx().judgeFailRetryLimit,
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
codexThreadId: task.codexThreadId,
activeTurnId: task.activeTurnId,
lastError: task.lastError,
@@ -314,6 +318,7 @@ function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTa
currentMode: task.currentMode,
judgeFailCount: task.judgeFailCount,
judgeFailRetryLimit: ctx().judgeFailRetryLimit,
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
codexThreadId: task.codexThreadId,
activeTurnId: task.activeTurnId,
lastError: task.lastError,
@@ -410,6 +415,18 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
const activeTaskIds = Array.from(activeTaskIdSet).sort();
const activeTaskId = activeTaskIds[0] ?? null;
const queues = perQueueSummaries(tasks, queueRecords);
const executionDiagnostics = buildExecutionDiagnostics({
controlPlane: "D601-code-queue-scheduler",
executionStateSource: "scheduler-execution-plane",
tasks,
activeRuns: ctx().activeRuns.values(),
activeRunSlotCount: activeRunSlots.length,
activeQueueIds: activeRunSlots,
processingQueueIds: Array.from(ctx().processingQueues),
orphanedActiveTaskIds,
oaPublisher: ctx().oaEventPublisherStatus(),
heartbeatStaleMs: schedulerHeartbeatStaleMs,
});
const summary: Record<string, JsonValue> = {
total: tasks.length,
defaultQueueId: ctx().defaultQueueId,
@@ -424,6 +441,8 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
orphanedActiveTaskIds,
orphanedActiveTaskCount: orphanedActiveTaskIds.length,
processing: ctx().processing(),
executionDiagnostics: executionDiagnostics as unknown as JsonValue,
schedulerHeartbeatStaleMs,
counts,
unreadTerminal,
judgeConfigured: ctx().config.minimaxApiKey.length > 0,
@@ -489,6 +508,7 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
lastError: ctx().codexSqliteLogExporter().lastError,
},
},
reconcile: ctx().reconcileStatus(tasks),
};
if (includeDevReady) summary.devReady = ctx().collectDevReady();
return summary;
src/components/microservices/code-queue/src/task-view.ts
+3
View File
@@ -1463,6 +1463,7 @@ function taskForMetaResponse(task: QueueTask): JsonValue {
judgeFailRetryLimit,
codexThreadId: task.codexThreadId,
activeTurnId: task.activeTurnId,
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
finalResponse: task.finalResponse,
lastError: task.lastError,
lastJudge: task.lastJudge,
@@ -1533,6 +1534,7 @@ function taskForCompactMetaResponse(task: QueueTask): JsonValue {
judgeFailRetryLimit,
codexThreadId: task.codexThreadId,
activeTurnId: task.activeTurnId,
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
finalResponse: prefixPreview(task.finalResponse, 1200),
lastError: task.lastError,
lastJudge: task.lastJudge,
@@ -2312,6 +2314,7 @@ function taskTraceSummaryResponse(task: QueueTask, oaTraceStats: JsonValue | nul
updatedAt: task.updatedAt,
currentAttempt: task.currentAttempt,
maxAttempts: task.maxAttempts,
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
stepCount,
llmStepCount,
promptEditable: ctx().queuedTaskPromptEditable(task),
src/components/microservices/code-queue/src/types.ts
+64
View File
@@ -30,6 +30,69 @@ export type TerminalStatus = "completed" | "interrupted" | "failed" | null;
export type TranscriptKind = "ran" | "explored" | "edited" | "plan" | "message" | "system" | "error";
export type CodeQueueExecutionPlane = "scheduler-execution-plane" | "postgres-control-plane";
export type CodeQueueExecutionHealth = "healthy" | "degraded" | "split-brain" | "stale-active";
export type CodeQueueRecoveryCandidateReason =
| "scheduler-local-active"
| "owner-heartbeat-fresh"
| "owner-heartbeat-expired"
| "owner-heartbeat-missing"
| "not-db-active";
export interface SchedulerActiveRunHeartbeat {
taskId: string;
queueId: string;
attempt: number;
activeTurnId: string | null;
codexThreadId: string | null;
owner: string;
schedulerInstance: string;
executionPlane: "scheduler-execution-plane";
agentPort: string;
status: TaskStatus;
lastLocalHeartbeatAt: string;
lastObservedAgentEventAt: string | null;
lastPersistedTraceAt: string | null;
outputMaxSeq: number;
source: "scheduler";
}
export interface CodeQueueExecutionDiagnostics {
state: CodeQueueExecutionHealth;
health: CodeQueueExecutionHealth;
degraded: boolean;
splitBrain: boolean;
executionStateSource: CodeQueueExecutionPlane;
controlPlane: string;
databaseActiveTaskIds: string[];
databaseActiveTaskCount: number;
schedulerActiveTaskIds: string[];
schedulerActiveTaskCount: number;
schedulerActiveRunSlotCount: number;
schedulerActiveQueueIds: string[];
schedulerProcessingQueueIds: string[];
schedulerOrphanedActiveTaskIds: string[];
schedulerOrphanedActiveTaskCount: number;
activeHeartbeatTaskIds: string[];
activeHeartbeatCount: number;
heartbeatFreshTaskIds: string[];
heartbeatExpiredTaskIds: string[];
heartbeatMissingTaskIds: string[];
staleRecoveryCandidateTaskIds: string[];
traceGapTaskIds: string[];
traceGapNotStaleTaskIds: string[];
schedulerHeartbeatStaleMs: number;
now: string;
lastSchedulerHeartbeatAt: string | null;
lastObservedAgentEventAt: string | null;
lastPersistedTraceAt: string | null;
oaPublisher: JsonValue | null;
reasons: string[];
guidance: string[];
}
export interface RuntimeConfig {
host: string;
port: number;
@@ -330,6 +393,7 @@ export interface QueueTask {
currentMode: RunMode | null;
codexThreadId: string | null;
activeTurnId: string | null;
schedulerHeartbeat?: SchedulerActiveRunHeartbeat | null;
finalResponse: string;
stepCount?: number;
llmStepCount?: number;
src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json
+80 -2
View File
@@ -16,7 +16,8 @@
"route": {
"kind": "kubernetes-service",
"serviceName": "code-queue-scheduler",
"servicePort": 4222
"servicePort": 4222,
"deploymentName": "code-queue"
},
"activeInstanceId": "D601",
"singleWriter": true,
@@ -127,7 +128,8 @@
"route": {
"kind": "kubernetes-service",
"serviceName": "code-queue-scheduler",
"servicePort": 4222
"servicePort": 4222,
"deploymentName": "code-queue"
},
"activeInstanceId": "D601-scheduler",
"singleWriter": true,
@@ -146,5 +148,81 @@
],
"requireAllInstancesHealthy": false
}
},
{
"apiVersion": "unidesk.ai/k3s/v1",
"kind": "ManagedKubernetesService",
"metadata": {
"name": "d601-provider-egress-proxy",
"namespace": "unidesk"
},
"spec": {
"adapterServiceId": "k3sctl-adapter",
"controlPlane": {
"type": "kubernetes",
"cluster": "unidesk-k3s",
"context": "unidesk-k3s"
},
"route": {
"kind": "kubernetes-service",
"serviceName": "d601-provider-egress-proxy",
"servicePort": 18789,
"deploymentName": "d601-provider-egress-proxy"
},
"activeInstanceId": "D601-provider-egress",
"singleWriter": false,
"expectedNodeIds": [
"D601"
],
"instances": [
{
"id": "D601-provider-egress",
"nodeId": "D601",
"role": "primary",
"baseUrl": "kubernetes://unidesk/services/d601-provider-egress-proxy:18789",
"healthPath": "/__unidesk/egress-proxy/health",
"healthMode": "pod-ready"
}
],
"requireAllInstancesHealthy": false
}
},
{
"apiVersion": "unidesk.ai/k3s/v1",
"kind": "ManagedKubernetesService",
"metadata": {
"name": "d601-tcp-egress-gateway",
"namespace": "unidesk"
},
"spec": {
"adapterServiceId": "k3sctl-adapter",
"controlPlane": {
"type": "kubernetes",
"cluster": "unidesk-k3s",
"context": "unidesk-k3s"
},
"route": {
"kind": "kubernetes-service",
"serviceName": "d601-tcp-egress-gateway",
"servicePort": 18080,
"deploymentName": "d601-tcp-egress-gateway"
},
"activeInstanceId": "D601-tcp-egress",
"singleWriter": false,
"expectedNodeIds": [
"D601"
],
"instances": [
{
"id": "D601-tcp-egress",
"nodeId": "D601",
"role": "primary",
"baseUrl": "kubernetes://unidesk/services/d601-tcp-egress-gateway:18080",
"healthPath": "/health",
"healthMode": "service-proxy"
}
],
"requireAllInstancesHealthy": false
}
}
]
src/components/microservices/k3sctl-adapter/src/index.ts
+73 -1
View File
@@ -430,6 +430,10 @@ function nativeServiceRef(service: ManagedService): { namespace: string; service
};
}
function deploymentName(service: ManagedService): string {
return routeString(service, "deploymentName", service.id);
}
function serviceProxyApiPath(service: ManagedService, targetPath: string): string {
const { serviceName, servicePort } = nativeServiceRef(service);
const safeTargetPath = targetPath.startsWith("/") ? targetPath : `/${targetPath}`;
@@ -922,6 +926,7 @@ async function probeEndpoint(endpoint: ManagedEndpoint): Promise<JsonRecord> {
async function probeKubernetesServiceActive(service: ManagedService): Promise<JsonRecord> {
const endpoint = activeEndpoint(service);
if (endpoint.healthMode === "pod-ready") return await probeKubernetesPodReady(service, endpoint);
return probeKubernetesEndpoint(service, endpoint, true);
}
@@ -995,6 +1000,68 @@ function podSummary(item: unknown): JsonRecord {
};
}
function conditionTrue(item: unknown, type: string): boolean {
const conditions = jsonAtPath(item, "status.conditions");
return Array.isArray(conditions) && conditions.some((condition) => {
const record = typeof condition === "object" && condition !== null ? condition as Record<string, unknown> : {};
return record.type === type && record.status === "True";
});
}
function integerAtPath(item: unknown, path: string): number {
const value = jsonAtPath(item, path);
return typeof value === "number" && Number.isFinite(value) ? Math.floor(value) : 0;
}
async function deploymentAvailabilityCheck(service: ManagedService): Promise<JsonRecord> {
const name = deploymentName(service);
const namespace = service.namespace;
try {
const deployment = await kubeApiJson(`/apis/apps/v1/namespaces/${encodeURIComponent(namespace)}/deployments/${encodeURIComponent(name)}`, config.healthTimeoutMs);
const desiredReplicas = Math.max(1, integerAtPath(deployment, "spec.replicas"));
const availableReplicas = integerAtPath(deployment, "status.availableReplicas");
const readyReplicas = integerAtPath(deployment, "status.readyReplicas");
const updatedReplicas = integerAtPath(deployment, "status.updatedReplicas");
const unavailableReplicas = integerAtPath(deployment, "status.unavailableReplicas");
const available = conditionTrue(deployment, "Available") || availableReplicas >= desiredReplicas;
return {
ok: available,
name,
namespace,
available,
desiredReplicas,
availableReplicas,
readyReplicas,
updatedReplicas,
unavailableReplicas,
};
} catch (error) {
return { ok: false, name, namespace, available: false, error: errorToJson(error) };
}
}
async function endpointNonEmptyCheck(service: ManagedService): Promise<JsonRecord> {
const { namespace, serviceName } = nativeServiceRef(service);
try {
const endpoints = await kubeApiJson(`/api/v1/namespaces/${encodeURIComponent(namespace)}/endpoints/${encodeURIComponent(serviceName)}`, config.healthTimeoutMs);
const subsets = Array.isArray(endpoints.subsets) ? endpoints.subsets : [];
const addresses = subsets.flatMap((subset) => {
const record = typeof subset === "object" && subset !== null ? subset as Record<string, unknown> : {};
return Array.isArray(record.addresses) ? record.addresses : [];
});
const readyAddressCount = addresses.length;
return {
ok: readyAddressCount > 0,
serviceName,
namespace,
readyAddressCount,
subsetCount: subsets.length,
};
} catch (error) {
return { ok: false, serviceName, namespace, readyAddressCount: 0, error: errorToJson(error) };
}
}
async function probeKubernetesPodReady(service: ManagedService, endpoint: ManagedEndpoint): Promise<JsonRecord> {
const checkedAt = new Date().toISOString();
const { namespace } = kubernetesEndpointServiceRef(service, endpoint);
@@ -1195,6 +1262,8 @@ async function serviceDiagnostics(service: ManagedService): Promise<JsonRecord>
status: "unhealthy",
error: errorToJson(error),
} satisfies JsonRecord));
const deployment = await deploymentAvailabilityCheck(service);
const endpoint = await endpointNonEmptyCheck(service);
const kubernetesApiServiceProxyOk = kubernetesApiServiceProxy.ok === true;
const targetServiceOk = targetService.ok === true;
const checks = {
@@ -1212,8 +1281,11 @@ async function serviceDiagnostics(service: ManagedService): Promise<JsonRecord>
},
targetService,
managedService,
deployment,
endpoint,
} satisfies Record<string, JsonValue>;
const ok = kubernetesApiServiceProxyOk && targetServiceOk;
const managedServiceOk = managedService.ok === true;
const ok = kubernetesApiServiceProxyOk && (targetServiceOk || managedServiceOk) && deployment.ok === true && endpoint.ok === true;
return {
ok,
service: "k3sctl-adapter",