fix: harden pgdata backup scheduler recovery

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")
+        );
+    }
+}