diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 2857362e..d1c88691 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -36,6 +36,7 @@ CLI 可以从 `master` 快速演进，但必须兼容 `deploy.json` 固定的 CI - `gh issue edit 24 --body-file --notify-claudeqq-brief-diff [--dry-run]` 是指挥简报 #24 的通知入口。正式执行会先读取 GitHub 上 #24 旧正文并通过 #24 body profile guard，再从 `--body-file` 读取新正文；随后先 PATCH issue 主体，再把本次新增的 `## 更新 YYYY-MM-DD HH:MM 北京时间` 段落发送给 ClaudeQQ，ClaudeQQ 失败不会回滚 issue 正文，失败只体现在返回 JSON 的 `claudeqq.ok=false` 和结构化 `degradedReason`。带通知 flag 的 `--dry-run` 不 PATCH、不发送；它按新正文做发送预览，并在输出中标明非 dry-run 才会读取旧正文做可靠 diff。默认 ClaudeQQ 目标是私聊 `645275593`，默认 base URL 是 UniDesk 受控入口 `http://backend-core:8080/api/microservices/claudeqq/proxy`，可用 `UNIDESK_COMMANDER_BRIEF_CLAUDEQQ_ENABLED`、`UNIDESK_COMMANDER_BRIEF_CLAUDEQQ_BASE_URL`、`UNIDESK_COMMANDER_BRIEF_CLAUDEQQ_TARGET_TYPE`、`UNIDESK_COMMANDER_BRIEF_CLAUDEQQ_USER_ID`、`UNIDESK_COMMANDER_BRIEF_CLAUDEQQ_GROUP_ID` 和 `UNIDESK_COMMANDER_BRIEF_CLAUDEQQ_TIMEOUT_MS` 覆盖。 - `gh issue scan-escape [--repo owner/name] [--limit N]` 只读扫描 issue 主体和 comments 中的字面量 `\n`、可疑 `\t`、shell newline escape 和 ANSI escape 字符串，输出 issue/comment id、url、kind、snippet，不自动修复。`gh pr list|view [--json ...]` 提供 REST 列表和详情，PR 字段白名单是 `body,title,state,number,url,author,head,base,draft,createdAt,updatedAt`。`gh pr create --title --body-file <file>|--body <text> --base <branch> --head <branch> [--draft] [--dry-run]`、`gh pr update <number> --mode replace|append --body-file <file>|--body <text> [--title ...] [--dry-run]`、`gh pr comment create <number> --body-file <file>|--body <text> [--dry-run]`、`gh pr comment delete <commentId> [--dry-run]`、`gh pr close|reopen <number> [--dry-run]` 是 PR CRUD/生命周期入口。`pr create --dry-run` 只输出 planned operation，不访问 GitHub；非 dry-run 创建前会校验 repo、base、head 和 compare ahead 状态，成功时返回 PR number/url。`pr update --mode append` 会先读取当前 PR body 再追加正文。`gh pr delete <number>` 和 `gh pr merge` 本阶段不开放，始终结构化返回 `unsupported-command`；PR 生命周期删除语义请使用 `close`。 - `ci install|status|run|publish-backend-core|publish-user-service|run-dev-e2e|logs` 管理 D601 原生 k3s 上的 Tekton CI。`run` 手动创建每 commit 检查和 Code Queue 只读性能门禁；`publish-backend-core` 与 `publish-user-service` 从 pushed Git commit 构建并发布 `127.0.0.1:5000/unidesk/<service>:<commit>` commit-pinned artifacts，输出 `artifactSummary`（含 `serviceId`、`sourceCommit`、`sourceRepo`、`dockerfile`、`imageRef`、`tag`、`digest`、`digestRef`），但不部署生产；`run-dev-e2e` 的 Git 控制 runner、短 launcher、host fetch 边界、临时 smoke namespace 和 no-CD 规则只在 `docs/reference/dev-ci-runner.md` 定义；Tekton CI 通用规则见 `docs/reference/ci.md`。 +- `schedule list|get|runs|run|retry-run|delete|upsert-pgdata-backup` 管理 backend-core 定时任务和运行历史。`schedule runs --limit N` 是全局历史视图，返回 `scope=global` 和 `scheduleId=null`；`schedule runs <scheduleId> --limit N` 是指定 schedule 历史视图，返回 `scope=schedule` 和对应 `scheduleId`。CLI 必须拒绝 `schedule runs 50` 这类纯数字位置参数，并提示使用 `schedule runs --limit 50`，避免把空数组误判成“没有历史 run”。`schedule run <id> --wait-ms N` 触发同一 schedule，并且即使 wait 超时也必须返回 `newRunId` 和 `observeCommand`；`schedule retry-run <failedRunId>` 只接受 failed run，从原 run 反查 `scheduleId` 后重触发同一 schedule，并输出 `originalRunId`、`scheduleId`、`newRunId` 和 `observeCommand`。 - `codex deploy <commitId>` 是旧 Code Queue 兼容部署入口，已禁用以防止维护通道直连 D601 部署 Code Queue；当前 dev 自动化只做 `ci run-dev-e2e` smoke，不提供 Code Queue CD，详细规则见 `docs/reference/codex-deploy.md`。 - `codex submit [prompt] [--prompt-file path|--prompt-stdin] [--queue queueId] [--provider-id id] [--cwd path] [--model model] [--reasoning-effort effort] [--execution-mode mode] [--max-attempts N] [--reference-task-id id] [--dry-run]` 通过 backend-core 私有代理向稳定 `code-queue` 用户服务路径提交任务；prompt 必须且只能来自位置参数、文件或 stdin 之一，`--dry-run` 只返回结构化请求且不实际入队。提交确认和 dry-run 必须返回完整 prompt、字符数和 `truncated=false`，不能套用任务详情的预览截断策略，否则长任务 prompt 无法被人工验收。真实提交会经过本机本地串行化保护和短节流，避免同一指挥端并发 submit 把低内存主机或 `code-queue-mgr` 控制面打抖；返回值会附带 `submitConcurrencyGuard` 说明本次提交的锁与等待信息。backend-core 默认把提交、队列 CRUD、已读状态、历史摘要和轻量 Trace 读取分流到主 server `code-queue-mgr`，由它写入主 PostgreSQL；D601 scheduler 只轮询并执行已入库任务。 - `codex task <taskId>` 通过 Code Queue 私有代理按任务 ID 查询结构化执行摘要；默认只返回有界 prompt/response 预览、执行 Provider、工作目录、最后 assistant message、最近工具调用摘要、attempt、judge、错误、耗时和 trace 翻页提示，适合在新队列任务中引用历史 session 且避免噪声爆炸。该摘要读取默认由主 server `code-queue-mgr` 从 PostgreSQL 返回，不依赖 D601 `code-queue-read` Service 可用。 diff --git a/docs/reference/microservices.md b/docs/reference/microservices.md index f5e51ba9..d9704e33 100644 --- a/docs/reference/microservices.md +++ b/docs/reference/microservices.md @@ -126,6 +126,9 @@ Project Manager 的标准发布是 `bun scripts/cli.ts ci publish-user-service - - 配置密钥：Compose 只透传 `UNIDESK_BAIDU_NETDISK_CLIENT_ID`、`UNIDESK_BAIDU_NETDISK_CLIENT_SECRET`、`UNIDESK_BAIDU_NETDISK_TOKEN_KEY` 与可选 `UNIDESK_BAIDU_NETDISK_APP_ROOT`；当前默认工作根目录为 `/`，如需收回到应用目录可显式设为 `/apps/<name>`；不得把百度 AppSecret、token key、access token 或 refresh token 写入仓库文件。 - artifact CD 密钥与健康门禁：dev/prod `deploy apply` 或 `artifact-registry deploy-service` 在 live apply 前必须从 canonical `.state/docker-compose.env` 确认 `UNIDESK_BAIDU_NETDISK_CLIENT_ID`、`UNIDESK_BAIDU_NETDISK_CLIENT_SECRET`、`UNIDESK_BAIDU_NETDISK_TOKEN_KEY` 存在，只能输出 present/length/boolean，不得打印值。单服务 recreate 后必须确认 `/health.auth.configured`、`clientIdConfigured`、`clientSecretConfigured`、`tokenKeyConfigured`、`loggedIn` 全部为 true；否则发布不能视为成功，需先恢复 env 注入并重新执行受控单服务 recreate。 - 配置步骤：`UNIDESK_BAIDU_NETDISK_TOKEN_KEY` 可由本机生成；百度 `client_id` 和 `client_secret` 必须由账号拥有者在百度网盘开放平台创建应用后提供，操作清单见 `docs/issue/baidu-netdisk-env-setup.md`。该环境配置说明只用于密钥、登录和维护验证；按 commit 发布或恢复 desired state 时仍必须使用上面的 CI artifact 加 dev/prod `deploy apply` 路径，不能把本地 `server rebuild baidu-netdisk` 当作镜像化交付证据。 +- PGDATA 备份预检：`schedule upsert-pgdata-backup` 创建的 PGDATA 到百度网盘日备份在执行 `pg_basebackup`、打包和上传前必须先做轻量 preflight。preflight 至少检查 `baidu-netdisk /health`、`clientIdConfigured`、`clientSecretConfigured`、`tokenKeyConfigured`、`loggedIn=true`、共享 staging 目录可写，以及 backend-core 到 PostgreSQL 的基本 `select 1` 可达性。preflight 失败必须快速结束 run，不得生成大备份文件或启动上传；run `result` 必须包含 `stage=preflight`、`failureKind`、`retryable`、`recommendedAction`、`observedAt` 和分项 `checks`。 +- PGDATA 备份失败分类：`failureKind=config-missing` 表示 Baidu Netdisk client id/client secret/token key 缺失，通常不可直接重试，先恢复 runtime env 并验证 `microservice health baidu-netdisk`；`auth-missing` 表示未登录或登录态不可用，先在 UniDesk Baidu Netdisk 页面重新登录；`staging-unwritable` 表示 `/data/baidu-netdisk-staging` 对 backend-core 不可写或路径配置错误，先修复挂载/权限；`database-unreachable` 表示 backend-core 到 PostgreSQL 路由或凭据不可用，先修复数据库可达性；`baidu-unreachable` 表示 Baidu Netdisk 服务健康端点不可达或服务自身不健康，先恢复服务健康；`unknown` 只用于未归类异常。排障时先用 `bun scripts/cli.ts schedule get unidesk-pgdata-baidu-daily` 或 `bun scripts/cli.ts schedule runs unidesk-pgdata-baidu-daily --limit 5` 查看最近 run 的 `result.failureKind`，不要根据全局空数组判断没有历史。 +- PGDATA 备份重试入口：失败 run 修复后使用 `bun scripts/cli.ts schedule retry-run <failedRunId>`，CLI 会输出 `originalRunId`、原 `scheduleId`、新 `newRunId` 和 `observeCommand`；也可用 `bun scripts/cli.ts schedule run unidesk-pgdata-baidu-daily --wait-ms <N>` 手动触发，同样会输出新 run id。若 `--wait-ms` 超时，继续执行返回的 `observeCommand` 观察，不要手工拼 backend-core API 路径。 - 数据库：OAuth 设备码会话、账号摘要、加密 token、传输任务和事件写入主 PostgreSQL 表 `baidu_netdisk_*`；服务启动时自动创建/补齐 schema，不依赖仅首次生效的 database init SQL。 - 文件边界：v1 只支持容器 staging 目录 `/data/staging` 与百度网盘配置工作根之间的后台上传/下载任务；staging 目录由主 server `.state/baidu-netdisk/staging` 挂载，`.state/` 只保存可重建文件缓存，不作为 token 或任务权威状态。当前授权账号已实测可对百度网盘根目录 `/` 执行列表、上传、获取 dlink、下载和删除临时探针，因此 `UNIDESK_BAIDU_NETDISK_APP_ROOT` 默认直接设为 `/`；仅当该值配置为 `/apps/...` 时，后端才会确保应用目录存在，目录已存在时必须返回/记录 `errno=-8` 并继续，禁止使用会重命名的策略重复创建 `_YYYYMMDD_...` 目录。 - API：`GET /health`；`GET /api/auth/status`；`POST /api/auth/device/start`；`GET /api/auth/device/status`；`POST /api/auth/refresh`；`POST /api/auth/logout`；`GET /api/account`；`GET /api/files`；`GET /api/files/meta`；`POST /api/folders`；`POST /api/files/manage`；`POST /api/transfers/upload-from-path`；`POST /api/transfers/download-to-path`；`POST /api/self-test`；`GET /api/transfers`；`GET|POST /api/transfers/{id}/cancel|retry`；`GET /logs`。 diff --git a/scripts/schedule-cli-contract-test.ts b/scripts/schedule-cli-contract-test.ts new file mode 100644 index 00000000..dc4d04bd --- /dev/null +++ b/scripts/schedule-cli-contract-test.ts @@ -0,0 +1,60 @@ +import { scheduleRetryRunObservation, scheduleRunObservation, scheduleRunsScope } from "./src/schedules"; + +type JsonRecord = Record<string, unknown>; + +function assertCondition(condition: unknown, message: string, detail: JsonRecord = {}): void { + if (!condition) throw new Error(`${message}: ${JSON.stringify(detail)}`); +} + +export function runScheduleCliContract(): JsonRecord { + const global = scheduleRunsScope(["runs", "--limit", "50"]); + assertCondition(global.scheduleId === null, "global schedule runs must not treat --limit value as schedule id", global); + assertCondition(global.limit === 50, "global schedule runs limit should be preserved", global); + + const scoped = scheduleRunsScope(["runs", "unidesk-pgdata-baidu-daily", "--limit", "5"]); + assertCondition(scoped.scheduleId === "unidesk-pgdata-baidu-daily", "schedule-specific runs should preserve schedule id", scoped); + assertCondition(scoped.limit === 5, "schedule-specific runs limit should be preserved", scoped); + + let numericScheduleRejected = false; + try { + scheduleRunsScope(["runs", "50"]); + } catch (error) { + numericScheduleRejected = String((error as Error).message).includes("schedule runs --limit N"); + } + assertCondition(numericScheduleRejected, "numeric positional schedule id should point operators to global --limit syntax"); + + const timeoutObservation = scheduleRunObservation( + "unidesk-pgdata-baidu-daily", + { body: { run: { id: "schedrun_new" } } }, + { ok: false, timedOut: true, timeoutMs: 1 }, + ); + assertCondition(timeoutObservation.newRunId === "schedrun_new", "schedule run output must expose newRunId even when wait times out", timeoutObservation); + assertCondition(String(timeoutObservation.observeCommand).includes("schedule runs unidesk-pgdata-baidu-daily --limit 20"), "schedule run output must expose observeCommand", timeoutObservation); + + const retryObservation = scheduleRetryRunObservation("schedrun_failed", { + body: { + originalRunId: "schedrun_failed", + scheduleId: "unidesk-pgdata-baidu-daily", + newRunId: "schedrun_retry", + }, + }); + assertCondition(retryObservation.originalRunId === "schedrun_failed", "retry-run output must preserve originalRunId", retryObservation); + assertCondition(retryObservation.scheduleId === "unidesk-pgdata-baidu-daily", "retry-run output must expose scheduleId", retryObservation); + assertCondition(retryObservation.newRunId === "schedrun_retry", "retry-run output must expose newRunId", retryObservation); + assertCondition(String(retryObservation.observeCommand).includes("schedule runs unidesk-pgdata-baidu-daily --limit 20"), "retry-run output must expose observeCommand", retryObservation); + + return { + ok: true, + checks: [ + "global schedule runs limit parsing", + "schedule-specific runs parsing", + "numeric positional guard", + "run wait timeout observation", + "retry-run observation", + ], + }; +} + +if (import.meta.main) { + process.stdout.write(`${JSON.stringify(runScheduleCliContract(), null, 2)}\n`); +} diff --git a/scripts/src/check.ts b/scripts/src/check.ts index 1f1cbe36..122c4001 100644 --- a/scripts/src/check.ts +++ b/scripts/src/check.ts @@ -284,6 +284,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default fileItem("scripts/src/ci.ts"), fileItem("scripts/src/e2e.ts"), fileItem("scripts/code-queue-prompt-observation-test.ts"), + fileItem("scripts/schedule-cli-contract-test.ts"), fileItem("scripts/src/artifact-registry.ts"), fileItem("src/components/microservices/k3sctl-adapter/k3s/ci/unidesk-ci.pipeline.yaml"), ); @@ -301,6 +302,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default items.push(commandItem("code-queue:control-plane-split-brain-diagnostics", ["bun", "scripts/code-queue-liveness-diagnostics-test.ts", "--only", "code-queue:control-plane-split-brain-diagnostics"], 30_000)); items.push(commandItem("code-queue:oa-publisher-degraded-visible", ["bun", "scripts/code-queue-liveness-diagnostics-test.ts", "--only", "code-queue:oa-publisher-degraded-visible"], 30_000)); items.push(commandItem("baidu-netdisk:artifact-guard-contract", ["bun", "scripts/baidu-netdisk-artifact-guard-contract-test.ts"], 30_000)); + items.push(commandItem("schedule:cli-contract", ["bun", "scripts/schedule-cli-contract-test.ts"], 30_000)); } else { items.push(skippedItem("typescript:scripts", "scripts TypeScript typecheck is opt-in", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:prompt-observation-contract", "prompt observation contract is opt-in with script checks", "--scripts-typecheck or --full")); @@ -308,6 +310,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default items.push(skippedItem("code-queue:trace-summary-contract", "Code Queue trace summary contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:liveness-diagnostics-fixtures", "Code Queue liveness diagnostics fixtures are opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("baidu-netdisk:artifact-guard-contract", "Baidu Netdisk artifact guard contract is opt-in with script checks", "--scripts-typecheck or --full")); + items.push(skippedItem("schedule:cli-contract", "Schedule CLI contract is opt-in with script checks", "--scripts-typecheck or --full")); } if (options.logs) { items.push(unifiedLogRotationItem()); diff --git a/scripts/src/help.ts b/scripts/src/help.ts index c6180fe7..4840a454 100644 --- a/scripts/src/help.ts +++ b/scripts/src/help.ts @@ -191,13 +191,14 @@ function scheduleHelp(): unknown { usage: [ "bun scripts/cli.ts schedule list", "bun scripts/cli.ts schedule get <id>", + "bun scripts/cli.ts schedule runs --limit N", "bun scripts/cli.ts schedule runs [scheduleId] [--limit N]", "bun scripts/cli.ts schedule run <id> [--wait-ms N]", "bun scripts/cli.ts schedule retry-run <failedRunId>", "bun scripts/cli.ts schedule delete <id>", "bun scripts/cli.ts schedule upsert-pgdata-backup [--time HH:MM] [--remote-base path]", ], - description: "Manage backend-core scheduled tasks and run history through the private core API.", + description: "Manage backend-core scheduled tasks and run history through the private core API. Global runs use schedule runs --limit N; schedule-specific runs pass a non-numeric schedule id.", }; } diff --git a/scripts/src/schedules.ts b/scripts/src/schedules.ts index cceafc2d..d91534d5 100644 --- a/scripts/src/schedules.ts +++ b/scripts/src/schedules.ts @@ -35,9 +35,40 @@ function terminalRunStatus(status: unknown): boolean { return ["succeeded", "failed", "skipped"].includes(String(status || "")); } +function isPlainNumeric(value: string): boolean { + return /^[0-9]+$/.test(value); +} + +export function scheduleRunsScope(args: string[]): { scheduleId: string | null; limit: number } { + const limit = numberOption(args, "--limit", 50); + let scheduleId: string | null = null; + for (let index = 1; index < args.length; index += 1) { + const arg = args[index]; + if (arg === "--limit") { + index += 1; + continue; + } + if (arg.startsWith("--")) { + throw new Error(`unsupported schedule runs option: ${arg}`); + } + if (scheduleId !== null) { + throw new Error("schedule runs accepts at most one schedule id"); + } + if (isPlainNumeric(arg)) { + throw new Error("schedule runs <scheduleId> requires a non-numeric schedule id; use schedule runs --limit N for the global history view"); + } + scheduleId = arg; + } + return { + scheduleId, + limit, + }; +} + async function waitForScheduleRun(scheduleId: string, runId: string, timeoutMs: number): Promise<unknown> { const started = Date.now(); let latest: unknown = null; + const observeCommand = `bun scripts/cli.ts schedule runs ${scheduleId} --limit 20`; while (Date.now() - started < timeoutMs) { latest = coreInternalFetch(`/api/schedules/${encodeURIComponent(scheduleId)}/runs?limit=20`); const runs = responseBody(latest).runs; @@ -47,7 +78,33 @@ async function waitForScheduleRun(scheduleId: string, runId: string, timeoutMs: } await Bun.sleep(2000); } - return { ok: false, timeoutMs, latest }; + return { ok: false, timedOut: true, timeoutMs, runId, scheduleId, observeCommand, latest }; +} + +export function scheduleRunObservation(scheduleId: string, response: unknown, wait: unknown): Record<string, unknown> { + const run = responseBody(response).run as { id?: unknown } | undefined; + const runId = typeof run?.id === "string" ? run.id : null; + return { + trigger: response, + originalRunId: run?.id ?? null, + scheduleId, + newRunId: runId, + observeCommand: `bun scripts/cli.ts schedule runs ${scheduleId} --limit 20`, + wait, + }; +} + +export function scheduleRetryRunObservation(originalRunId: string, response: unknown): Record<string, unknown> { + const body = responseBody(response); + const scheduleId = typeof body.scheduleId === "string" ? body.scheduleId : null; + const newRunId = typeof body.newRunId === "string" ? body.newRunId : null; + return { + originalRunId: String(body.originalRunId ?? originalRunId), + scheduleId, + newRunId, + observeCommand: scheduleId !== null && newRunId !== null ? `bun scripts/cli.ts schedule runs ${scheduleId} --limit 20` : null, + response, + }; } function pgdataBackupScheduleBody(config: UniDeskConfig, args: string[]): Record<string, unknown> { @@ -84,9 +141,9 @@ export async function runScheduleCommand(config: UniDeskConfig, args: string[]): return coreInternalFetch(`/api/schedules/${encodeURIComponent(idArg)}`); } if (action === "runs") { - const limit = numberOption(args, "--limit", 50); - return idArg - ? coreInternalFetch(`/api/schedules/${encodeURIComponent(idArg)}/runs?limit=${limit}`) + const { scheduleId, limit } = scheduleRunsScope(args); + return scheduleId + ? coreInternalFetch(`/api/schedules/${encodeURIComponent(scheduleId)}/runs?limit=${limit}`) : coreInternalFetch(`/api/schedules/runs?limit=${limit}`); } if (action === "delete") { @@ -99,12 +156,17 @@ export async function runScheduleCommand(config: UniDeskConfig, args: string[]): const run = responseBody(response).run as { id?: unknown } | undefined; const waitMs = numberOption(args, "--wait-ms", 0); const wait = waitMs > 0 && typeof run?.id === "string" ? await waitForScheduleRun(idArg, run.id, waitMs) : null; - return { trigger: response, wait }; + return scheduleRunObservation(idArg, response, wait); + } + if (action === "retry-run") { + if (!idArg) throw new Error("schedule retry-run requires failed run id"); + const response = coreInternalFetch(`/api/schedules/runs/${encodeURIComponent(idArg)}/retry`, { method: "POST", body: {} }); + return scheduleRetryRunObservation(idArg, response); } if (action === "upsert-pgdata-backup" || action === "pgdata-backup") { const body = pgdataBackupScheduleBody(config, args); const id = String(body.id || "unidesk-pgdata-baidu-daily"); return coreInternalFetch(`/api/schedules/${encodeURIComponent(id)}`, { method: "PUT", body }); } - throw new Error("schedule command must be one of: list, get, runs, run, delete, upsert-pgdata-backup"); + throw new Error("schedule command must be one of: list, get, runs, run, retry-run, delete, upsert-pgdata-backup"); } diff --git a/src/components/backend-core/src/scheduler.rs b/src/components/backend-core/src/scheduler.rs index 05b246e3..5bdaef03 100644 --- a/src/components/backend-core/src/scheduler.rs +++ b/src/components/backend-core/src/scheduler.rs @@ -449,7 +449,7 @@ fn safe_file_part(value: &str) -> String { fn database_connection_parts( database_url: &str, -) -> anyhow::Result<(String, String, String, String)> { +) -> anyhow::Result<(String, String, String, String, String)> { let url = url::Url::parse(database_url)?; Ok(( url.host_str().unwrap_or("database").to_string(), @@ -460,6 +460,14 @@ fn database_connection_parts( }, percent_decode(url.username()), url.password().map(percent_decode).unwrap_or_default(), + { + let name = percent_decode(url.path().trim_start_matches('/')); + if name.is_empty() { + "postgres".to_string() + } else { + name + } + }, )) } @@ -552,6 +560,222 @@ async fn baidu_json( Ok(value) } +async fn baidu_raw_json(base_url: &str, path: &str, timeout_ms: u64) -> anyhow::Result<Value> { + let client = reqwest::Client::new(); + let url = reqwest::Url::parse(base_url)?.join(path.trim_start_matches('/'))?; + fetch_json_with_timeout(path.to_string(), client.get(url), timeout_ms).await +} + +fn preflight_failure( + failure_kind: &str, + retryable: bool, + recommended_action: &str, + checks: Value, +) -> Value { + json!({ + "ok": false, + "stage": "preflight", + "failureKind": failure_kind, + "retryable": retryable, + "recommendedAction": recommended_action, + "observedAt": crate::json_util::now_iso(), + "checks": checks, + }) +} + +fn baidu_health_preflight_failure(health: &Value) -> Option<(&'static str, bool, &'static str)> { + if health.get("ok").and_then(Value::as_bool) == Some(false) { + return Some(( + "baidu-unreachable", + true, + "Check the baidu-netdisk service health and PostgreSQL schema readiness before retrying the backup.", + )); + } + let auth = health.get("auth").unwrap_or(&Value::Null); + let client_id_configured = auth + .get("clientIdConfigured") + .and_then(Value::as_bool) + .unwrap_or(false); + let client_secret_configured = auth + .get("clientSecretConfigured") + .and_then(Value::as_bool) + .unwrap_or(false); + let token_key_configured = auth + .get("tokenKeyConfigured") + .and_then(Value::as_bool) + .unwrap_or(false); + if !client_id_configured || !client_secret_configured || !token_key_configured { + return Some(( + "config-missing", + false, + "Restore BAIDU_NETDISK_CLIENT_ID, BAIDU_NETDISK_CLIENT_SECRET and BAIDU_NETDISK_TOKEN_KEY in the baidu-netdisk runtime environment, then verify microservice health baidu-netdisk.", + )); + } + if auth.get("loggedIn").and_then(Value::as_bool) != Some(true) { + return Some(( + "auth-missing", + false, + "Log in again from the UniDesk Baidu Netdisk page, then verify /api/auth/status reports loggedIn=true.", + )); + } + None +} + +async fn pgdata_backup_preflight( + state: &Arc<AppState>, + staging_subdir: &str, + baidu_base_url: &str, + db_host: &str, + db_port: &str, + db_user: &str, + db_password: &str, + db_name: &str, + run_id: &str, +) -> Value { + let mut checks = serde_json::Map::new(); + + let health = match baidu_raw_json(baidu_base_url, "/health", 10_000).await { + Ok(health) => health, + Err(error) => { + checks.insert( + "baidu".to_string(), + json!({ "ok": false, "error": error.to_string(), "baseUrl": baidu_base_url }), + ); + return preflight_failure( + "baidu-unreachable", + true, + "Check that baidu-netdisk is running on the internal Compose network and retry after the health endpoint is reachable.", + Value::Object(checks), + ); + } + }; + if let Some((failure_kind, retryable, recommended_action)) = + baidu_health_preflight_failure(&health) + { + checks.insert( + "baidu".to_string(), + json!({ "ok": false, "health": compact_json(&health) }), + ); + return preflight_failure( + failure_kind, + retryable, + recommended_action, + Value::Object(checks), + ); + } + checks.insert( + "baidu".to_string(), + json!({ "ok": true, "health": compact_json(&health) }), + ); + + let probe_relative_dir = format!("{staging_subdir}/.preflight_{run_id}"); + let (_, probe_dir) = match normalize_relative_staging_path( + &state.config.pgdata_backup_staging_dir, + &probe_relative_dir, + ) { + Ok(paths) => paths, + Err(error) => { + checks.insert( + "staging".to_string(), + json!({ "ok": false, "error": error.to_string() }), + ); + return preflight_failure( + "staging-unwritable", + false, + "Fix PGDATA backup staging path configuration so it stays inside PGDATA_BACKUP_STAGING_DIR.", + Value::Object(checks), + ); + } + }; + let probe_file = probe_dir.join("write-check.tmp"); + let staging_result = async { + fs::create_dir_all(&probe_dir).await?; + fs::write(&probe_file, b"ok").await?; + anyhow::Ok(()) + } + .await; + if let Err(error) = staging_result { + checks.insert( + "staging".to_string(), + json!({ "ok": false, "path": probe_dir.to_string_lossy(), "error": error.to_string() }), + ); + return preflight_failure( + "staging-unwritable", + false, + "Ensure the shared Baidu Netdisk staging directory is mounted and writable by backend-core before retrying.", + Value::Object(checks), + ); + } + let _ = fs::remove_file(&probe_file).await; + let _ = fs::remove_dir(&probe_dir).await; + checks.insert( + "staging".to_string(), + json!({ "ok": true, "path": probe_dir.to_string_lossy() }), + ); + + let args = vec![ + "-h".to_string(), + db_host.to_string(), + "-p".to_string(), + db_port.to_string(), + "-U".to_string(), + db_user.to_string(), + "-d".to_string(), + db_name.to_string(), + "-tAc".to_string(), + "select 1".to_string(), + ]; + match run_local_command( + "psql", + &args, + vec![ + ("PGPASSWORD", db_password.to_string()), + ("PGCONNECT_TIMEOUT", "5".to_string()), + ], + 10_000, + ) + .await + { + Ok((true, stdout, _stderr, exit_code, timed_out)) => { + checks.insert( + "database".to_string(), + json!({ "ok": true, "host": db_host, "port": db_port, "database": db_name, "exitCode": exit_code, "timedOut": timed_out, "stdout": truncate_text(&stdout, 200) }), + ); + } + Ok((false, stdout, stderr, exit_code, timed_out)) => { + checks.insert( + "database".to_string(), + json!({ "ok": false, "host": db_host, "port": db_port, "database": db_name, "exitCode": exit_code, "timedOut": timed_out, "stdout": truncate_text(&stdout, 500), "stderr": truncate_text(&stderr, 1000) }), + ); + return preflight_failure( + "database-unreachable", + true, + "Check database reachability and credentials from backend-core before retrying the PGDATA backup.", + Value::Object(checks), + ); + } + Err(error) => { + checks.insert( + "database".to_string(), + json!({ "ok": false, "host": db_host, "port": db_port, "database": db_name, "error": error.to_string() }), + ); + return preflight_failure( + "database-unreachable", + true, + "Check that PostgreSQL client tools and the database route are available inside backend-core.", + Value::Object(checks), + ); + } + } + + json!({ + "ok": true, + "stage": "preflight", + "observedAt": crate::json_util::now_iso(), + "checks": Value::Object(checks), + }) +} + async fn wait_for_baidu_transfer( state: &Arc<AppState>, base_url: &str, @@ -683,8 +907,23 @@ async fn execute_pgdata_backup_action( )?; let remote_dir = normalize_remote_dir(&format!("{remote_base_dir}/{month}"))?; let remote_path = normalize_remote_dir(&format!("{remote_dir}/{filename}"))?; - let (db_host, db_port, db_user, db_password) = + let (db_host, db_port, db_user, db_password, db_name) = database_connection_parts(&state.config.database_url)?; + let preflight = pgdata_backup_preflight( + state, + staging_subdir, + baidu_base_url, + &db_host, + &db_port, + &db_user, + &db_password, + &db_name, + run_id, + ) + .await; + if preflight.get("ok").and_then(Value::as_bool) != Some(true) { + return Ok((false, preflight)); + } if let Some(parent) = backup_path.parent() { fs::create_dir_all(parent).await?; } @@ -744,6 +983,7 @@ async fn execute_pgdata_backup_action( "remoteDir": remote_dir, "month": month, "timestampPrefix": stamp, + "preflight": preflight, "baiduTransferJobId": job_id, "baiduTransferStatus": uploaded_job.get("status").and_then(Value::as_str).unwrap_or(""), "baiduFsId": uploaded_job.get("fsId").and_then(Value::as_str).unwrap_or(""), @@ -795,6 +1035,20 @@ async fn get_scheduled_task_row( .map(schedule_task_from_row)) } +async fn get_scheduled_task_run_row( + state: &Arc<AppState>, + run_id: &str, +) -> anyhow::Result<Option<ScheduledTaskRunRow>> { + let client = state.pool.get().await?; + Ok(client + .query_opt( + "SELECT * FROM unidesk_scheduled_task_runs WHERE id = $1 LIMIT 1", + &[&run_id], + ) + .await? + .map(schedule_run_from_row)) +} + async fn execute_scheduled_run(state: Arc<AppState>, run_id: String) { { let mut active = state.active_scheduled_runs.lock().await; @@ -1105,6 +1359,49 @@ async fn trigger_scheduled_task( Ok(Some(schedule_run_view(&run))) } +async fn retry_scheduled_run( + state: &Arc<AppState>, + original_run_id: &str, +) -> anyhow::Result<crate::http::HttpResponse> { + let Some(original_run) = get_scheduled_task_run_row(state, original_run_id).await? else { + return Ok(json_response( + json!({ "ok": false, "error": format!("scheduled run not found: {original_run_id}") }), + 404, + )); + }; + if original_run.status != "failed" { + return Ok(json_response( + json!({ + "ok": false, + "error": "only failed scheduled runs can be retried through this endpoint", + "originalRun": schedule_run_view(&original_run), + }), + 409, + )); + } + let schedule_id = original_run.schedule_id.clone(); + match trigger_scheduled_task(state, &schedule_id, "retry").await? { + Some(run) => { + let new_run_id = run.get("id").and_then(Value::as_str).unwrap_or(""); + Ok(json_response( + json!({ + "ok": true, + "originalRunId": original_run_id, + "scheduleId": schedule_id, + "newRunId": new_run_id, + "originalRun": schedule_run_view(&original_run), + "run": run, + }), + 200, + )) + } + None => Ok(json_response( + json!({ "ok": false, "error": format!("scheduled task was not retried: {schedule_id}") }), + 409, + )), + } +} + pub async fn scheduled_task_route( state: &Arc<AppState>, method: Method, @@ -1134,9 +1431,24 @@ pub async fn scheduled_task_route( upsert_scheduled_task(state, body, None).await } ([only], "GET") if only == "runs" => Ok(json_response( - json!({ "ok": true, "runs": get_scheduled_task_runs(state, None, read_limit(&url, 100)).await? }), + json!({ "ok": true, "scope": "global", "scheduleId": Value::Null, "runs": get_scheduled_task_runs(state, None, read_limit(&url, 100)).await? }), 200, )), + ([scope, run_id], "GET") if scope == "runs" => { + let Some(run) = get_scheduled_task_run_row(state, run_id).await? else { + return Ok(json_response( + json!({ "ok": false, "error": format!("scheduled run not found: {run_id}") }), + 404, + )); + }; + Ok(json_response( + json!({ "ok": true, "run": schedule_run_view(&run) }), + 200, + )) + } + ([scope, run_id, action], "POST") if scope == "runs" && action == "retry" => { + retry_scheduled_run(state, run_id).await + } ([schedule_id], "GET") => { let Some(schedule) = get_scheduled_task_row(state, schedule_id).await? else { return Ok(json_response( @@ -1169,7 +1481,7 @@ pub async fn scheduled_task_route( } } ([schedule_id, action], "GET") if action == "runs" => Ok(json_response( - json!({ "ok": true, "runs": get_scheduled_task_runs(state, Some(schedule_id), read_limit(&url, 100)).await? }), + json!({ "ok": true, "scope": "schedule", "scheduleId": schedule_id, "runs": get_scheduled_task_runs(state, Some(schedule_id), read_limit(&url, 100)).await? }), 200, )), _ => Ok(json_response( @@ -1250,3 +1562,69 @@ pub async fn run_due_scheduled_tasks(state: &Arc<AppState>) -> anyhow::Result<() } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn baidu_health_preflight_failure_classifies_missing_token_config() { + let health = json!({ + "ok": true, + "auth": { + "clientIdConfigured": true, + "clientSecretConfigured": true, + "tokenKeyConfigured": false, + "loggedIn": true + } + }); + let failure = baidu_health_preflight_failure(&health).expect("missing token must fail"); + assert_eq!(failure.0, "config-missing"); + assert!(!failure.1); + assert!(failure.2.contains("BAIDU_NETDISK_TOKEN_KEY")); + } + + #[test] + fn baidu_health_preflight_failure_classifies_logged_out_auth() { + let health = json!({ + "ok": true, + "auth": { + "clientIdConfigured": true, + "clientSecretConfigured": true, + "tokenKeyConfigured": true, + "loggedIn": false + } + }); + let failure = baidu_health_preflight_failure(&health).expect("logged out must fail"); + assert_eq!(failure.0, "auth-missing"); + assert!(!failure.1); + } + + #[test] + fn preflight_failure_exposes_structured_fields() { + let failure = preflight_failure( + "config-missing", + false, + "restore runtime env", + json!({ "baidu": { "ok": false } }), + ); + assert_eq!(failure.get("ok").and_then(Value::as_bool), Some(false)); + assert_eq!( + failure.get("stage").and_then(Value::as_str), + Some("preflight") + ); + assert_eq!( + failure.get("failureKind").and_then(Value::as_str), + Some("config-missing") + ); + assert_eq!( + failure.get("retryable").and_then(Value::as_bool), + Some(false) + ); + assert!(failure.get("observedAt").and_then(Value::as_str).is_some()); + assert_eq!( + failure.get("recommendedAction").and_then(Value::as_str), + Some("restore runtime env") + ); + } +}