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

fix: expose code queue stats buckets

Browse Source
This commit is contained in:
Codex
2026-05-20 07:41:39 +00:00
parent fadb40f744
commit fee1b1b710
1 changed files with 240 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src-rs/main.rs
+240 -4
View File
@@ -20,6 +20,7 @@ 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;
const CODEX_STATS_TIME_ZONE: &str = "Asia/Shanghai";
#[derive(Clone)]
struct Config {
@@ -86,6 +87,19 @@ struct TaskMeta {
task_json: Option<Value>,
}
#[derive(Clone, Debug)]
struct DailyTaskStatsRow {
date: String,
executed_tasks: i64,
completed_tasks: i64,
retry_attempts: i64,
succeeded_tasks: i64,
failed_tasks: i64,
canceled_tasks: i64,
total_duration_ms: i64,
duration_samples: i64,
}
#[derive(Clone, Debug)]
struct RequestUrl {
path: String,
@@ -593,6 +607,174 @@ fn timing_json(task: &TaskMeta) -> Value {
})
}
fn codex_stats_days_from_url(url: &RequestUrl) -> i64 {
parse_i64(query_param(url, "days"), 14).clamp(1, 90)
}
fn codex_stats_queue_id_from_url(url: &RequestUrl) -> Option<String> {
query_param(url, "queueId")
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.map(|value| safe_queue_id(&value))
}
fn codex_stats_range_sql() -> &'static str {
"
WITH stats_range AS (
SELECT
$2::int AS days,
to_char(now() AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS end_date,
(date_trunc('day', now() AT TIME ZONE 'Asia/Shanghai') - (($2::int - 1) * INTERVAL '1 day')) AT TIME ZONE 'Asia/Shanghai' AS start_at,
(date_trunc('day', now() AT TIME ZONE 'Asia/Shanghai') + INTERVAL '1 day') AT TIME ZONE 'Asia/Shanghai' AS end_at
),
day_series AS (
SELECT to_char(day::date, 'YYYY-MM-DD') AS date
FROM stats_range stats,
generate_series(
(stats.start_at AT TIME ZONE 'Asia/Shanghai')::date,
(stats.end_at AT TIME ZONE 'Asia/Shanghai')::date - 1,
INTERVAL '1 day'
) AS day
),
executed AS (
SELECT to_char(started_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS date, COUNT(*)::bigint AS count
FROM unidesk_code_queue_tasks, stats_range
WHERE ($1::text IS NULL OR queue_id = $1)
AND started_at >= stats_range.start_at
AND started_at < stats_range.end_at
GROUP BY date
),
completed AS (
SELECT
to_char(COALESCE(finished_at, updated_at) AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS date,
COUNT(*)::bigint AS completed_tasks,
COUNT(*) FILTER (WHERE status = 'succeeded')::bigint AS succeeded_tasks,
COUNT(*) FILTER (WHERE status = 'failed')::bigint AS failed_tasks,
COUNT(*) FILTER (WHERE status = 'canceled')::bigint AS canceled_tasks,
COALESCE(SUM(GREATEST(EXTRACT(EPOCH FROM (COALESCE(finished_at, updated_at) - COALESCE(started_at, created_at))) * 1000, 0)), 0)::bigint AS total_duration_ms,
COUNT(CASE WHEN COALESCE(started_at, created_at) IS NOT NULL THEN 1 END)::bigint AS duration_samples
FROM unidesk_code_queue_tasks, stats_range
WHERE ($1::text IS NULL OR queue_id = $1)
AND status IN ('succeeded', 'failed', 'canceled')
AND COALESCE(finished_at, updated_at) >= stats_range.start_at
AND COALESCE(finished_at, updated_at) < stats_range.end_at
GROUP BY date
),
retries AS (
SELECT
to_char(updated_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD') AS date,
SUM(GREATEST(current_attempt - 1, attempt_count - 1, 0))::bigint AS count
FROM unidesk_code_queue_tasks, stats_range
WHERE ($1::text IS NULL OR queue_id = $1)
AND updated_at >= stats_range.start_at
AND updated_at < stats_range.end_at
AND GREATEST(current_attempt - 1, attempt_count - 1, 0) > 0
GROUP BY date
)
SELECT
day_series.date,
COALESCE(executed.count, 0)::bigint AS executed_tasks,
COALESCE(completed.completed_tasks, 0)::bigint AS completed_tasks,
COALESCE(retries.count, 0)::bigint AS retry_attempts,
COALESCE(completed.succeeded_tasks, 0)::bigint AS succeeded_tasks,
COALESCE(completed.failed_tasks, 0)::bigint AS failed_tasks,
COALESCE(completed.canceled_tasks, 0)::bigint AS canceled_tasks,
COALESCE(completed.total_duration_ms, 0)::bigint AS total_duration_ms,
COALESCE(completed.duration_samples, 0)::bigint AS duration_samples,
(SELECT end_date FROM stats_range) AS end_date
FROM day_series
LEFT JOIN executed ON executed.date = day_series.date
LEFT JOIN completed ON completed.date = day_series.date
LEFT JOIN retries ON retries.date = day_series.date
ORDER BY day_series.date ASC
"
}
fn rounded_average_ms(total_duration_ms: i64, duration_samples: i64) -> Value {
if duration_samples <= 0 {
return Value::Null;
}
json!((total_duration_ms as f64 / duration_samples as f64).round() as i64)
}
fn codex_stats_summary_from_rows(days: i64, generated_at: &str, rows: &[DailyTaskStatsRow]) -> Value {
let mut total_executed = 0_i64;
let mut total_completed = 0_i64;
let mut total_retries = 0_i64;
let mut total_succeeded = 0_i64;
let mut total_failed = 0_i64;
let mut total_canceled = 0_i64;
let mut total_duration_ms = 0_i64;
let mut duration_samples = 0_i64;
let daily = rows
.iter()
.map(|row| {
total_executed += row.executed_tasks;
total_completed += row.completed_tasks;
total_retries += row.retry_attempts;
total_succeeded += row.succeeded_tasks;
total_failed += row.failed_tasks;
total_canceled += row.canceled_tasks;
total_duration_ms += row.total_duration_ms;
duration_samples += row.duration_samples;
json!({
"date": row.date,
"executedTasks": row.executed_tasks,
"completedTasks": row.completed_tasks,
"retryAttempts": row.retry_attempts,
"succeededTasks": row.succeeded_tasks,
"failedTasks": row.failed_tasks,
"canceledTasks": row.canceled_tasks,
"totalDurationMs": row.total_duration_ms,
"durationSamples": row.duration_samples,
"avgDurationMs": rounded_average_ms(row.total_duration_ms, row.duration_samples)
})
})
.collect::<Vec<_>>();
let start_date = rows.first().map(|row| row.date.clone()).unwrap_or_default();
let end_date = rows.last().map(|row| row.date.clone()).unwrap_or_default();
json!({
"generatedAt": generated_at,
"timezone": CODEX_STATS_TIME_ZONE,
"days": days,
"range": { "startDate": start_date, "endDate": end_date },
"totals": {
"executedTasks": total_executed,
"completedTasks": total_completed,
"retryAttempts": total_retries,
"succeededTasks": total_succeeded,
"failedTasks": total_failed,
"canceledTasks": total_canceled,
"totalDurationMs": total_duration_ms,
"durationSamples": duration_samples,
"avgDurationMs": rounded_average_ms(total_duration_ms, duration_samples)
},
"daily": daily
})
}
fn task_statistics_summary(state: &AppState, queue_id: Option<String>, days: i64) -> Result<Value, String> {
let mut client = db_client(state, "unidesk-code-queue-mgr-rust-stats")?;
let rows = client
.query(codex_stats_range_sql(), &[&queue_id, &days])
.map_err(|error| error.to_string())?;
let stats_rows = rows
.iter()
.map(|row| DailyTaskStatsRow {
date: row.get("date"),
executed_tasks: row.get("executed_tasks"),
completed_tasks: row.get("completed_tasks"),
retry_attempts: row.get("retry_attempts"),
succeeded_tasks: row.get("succeeded_tasks"),
failed_tasks: row.get("failed_tasks"),
canceled_tasks: row.get("canceled_tasks"),
total_duration_ms: row.get("total_duration_ms"),
duration_samples: row.get("duration_samples"),
})
.collect::<Vec<_>>();
Ok(codex_stats_summary_from_rows(days, &now_iso(), &stats_rows))
}
fn json_array_len(value: Option<&Value>, key: &str, fallback: i64) -> i64 {
value
.and_then(|json| json.get(key))
@@ -2316,8 +2498,15 @@ fn tasks_overview(state: &AppState, url: &RequestUrl) -> Result<(Value, u16), (V
let selected_enabled = query_param(url, "selected").as_deref() != Some("0");
let prefer_id = query_param(url, "preferId").unwrap_or_default();
let include_selected_json = selected_enabled && !prefer_id.is_empty();
let include_statistics = query_param(url, "stats").as_deref() != Some("0");
let (tasks, total, limit, offset) = load_tasks_for_list(state, url, false).map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
let summary = queue_summary(state).map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
let statistics = if include_statistics {
task_statistics_summary(state, codex_stats_queue_id_from_url(url), codex_stats_days_from_url(url))
.map_err(|error| (json!({ "ok": false, "error": error }), 500))?
} else {
Value::Null
};
let selected = if include_selected_json {
let selected_id = if prefer_id.is_empty() { tasks.first().map(|task| task.id.clone()).unwrap_or_default() } else { prefer_id };
if selected_id.is_empty() {
@@ -2358,7 +2547,7 @@ fn tasks_overview(state: &AppState, url: &RequestUrl) -> Result<(Value, u16), (V
json!({
"ok": true,
"queue": summary,
"statistics": { "days": 14, "source": "code-queue-mgr-rust", "skipped": true },
"statistics": statistics,
"tasks": tasks.iter().map(|task| task_list_response(state, task, true)).collect::<Vec<_>>(),
"selected": selected,
"pagination": {
@@ -2783,9 +2972,12 @@ fn handle_request(state: &AppState, request: &mut Request) -> (Value, u16) {
(Method::Post, "/api/queues") => return create_queue(state, request).unwrap_or_else(|response| response),
(Method::Post, "/api/queues/merge") => (json!({ "ok": false, "error": "queue merge requires /api/queues/:target/merge on rust mgr" }), 400),
(Method::Get, "/api/tasks/overview") => return tasks_overview(state, &url).unwrap_or_else(|response| response),
(Method::Get, "/api/tasks/stats") => match queue_summary(state) {
Ok(summary) => (json!({ "ok": true, "statistics": { "days": 14, "source": "code-queue-mgr-rust", "skipped": true }, "queue": summary }), 200),
Err(error) => (json!({ "ok": false, "error": error }), 500),
(Method::Get, "/api/tasks/stats") => match (
task_statistics_summary(state, codex_stats_queue_id_from_url(&url), codex_stats_days_from_url(&url)),
queue_summary(state),
) {
(Ok(statistics), Ok(summary)) => (json!({ "ok": true, "statistics": statistics, "queue": summary }), 200),
(Err(error), _) | (_, Err(error)) => (json!({ "ok": false, "error": error }), 500),
},
(Method::Get, "/api/tasks") => {
match load_tasks_for_list(state, &url, false).and_then(|(tasks, total, limit, offset)| {
@@ -3177,6 +3369,50 @@ mod tests {
assert!(sql.contains("'{nextMode}', to_jsonb('retry'::text)"), "{sql}");
}
#[test]
fn stats_summary_contract_exposes_daily_buckets_and_totals() {
let stats = codex_stats_summary_from_rows(
2,
"2026-05-20T00:00:00.000Z",
&[
DailyTaskStatsRow {
date: "2026-05-19".to_string(),
executed_tasks: 2,
completed_tasks: 1,
retry_attempts: 1,
succeeded_tasks: 1,
failed_tasks: 0,
canceled_tasks: 0,
total_duration_ms: 120_000,
duration_samples: 1,
},
DailyTaskStatsRow {
date: "2026-05-20".to_string(),
executed_tasks: 3,
completed_tasks: 2,
retry_attempts: 2,
succeeded_tasks: 1,
failed_tasks: 1,
canceled_tasks: 0,
total_duration_ms: 480_000,
duration_samples: 2,
},
],
);
assert_eq!(stats.get("timezone").and_then(Value::as_str), Some("Asia/Shanghai"));
assert_eq!(stats.pointer("/range/startDate").and_then(Value::as_str), Some("2026-05-19"));
assert_eq!(stats.pointer("/range/endDate").and_then(Value::as_str), Some("2026-05-20"));
assert_eq!(stats.pointer("/totals/executedTasks").and_then(Value::as_i64), Some(5));
assert_eq!(stats.pointer("/totals/completedTasks").and_then(Value::as_i64), Some(3));
assert_eq!(stats.pointer("/totals/retryAttempts").and_then(Value::as_i64), Some(3));
assert_eq!(stats.pointer("/totals/failedTasks").and_then(Value::as_i64), Some(1));
assert_eq!(stats.pointer("/totals/avgDurationMs").and_then(Value::as_i64), Some(200_000));
let daily = stats.get("daily").and_then(Value::as_array).expect("daily buckets");
assert_eq!(daily.len(), 2);
assert_eq!(daily[1].get("avgDurationMs").and_then(Value::as_i64), Some(240_000));
}
#[test]
fn manual_retry_canceled_task_becomes_scheduler_runnable() {
let state = test_state();