更新 main.rs

src/main.rs

@@ -11,15 +11,18 @@ use reqwest::StatusCode;
 use serde::{Deserialize, Serialize};
 use tokio::io::AsyncWriteExt;
 use tokio::process::{Child, Command};
-use tokio::time::{interval, timeout};
+use tokio::time::{interval, sleep, timeout};
 use walkdir::WalkDir;
 
-/// The OpenClaw config directory synced to/from HuggingFace.
 const WORKSPACE_DIR: &str = "/home/user/.openclaw";
-
 const STATE_FILE: &str = ".hf-sync-state.json";
 const FINAL_WAIT_TIMEOUT: Duration = Duration::from_secs(20);
 
+/// How long to wait for HF to become reachable at startup before giving up
+/// and continuing with an empty / local workspace.
+const NETWORK_STARTUP_TIMEOUT: Duration = Duration::from_secs(60);
+const NETWORK_RETRY_INTERVAL: Duration = Duration::from_secs(5);
+
 #[derive(Debug, Clone)]
 struct Config {
     token: String,
@@ -96,10 +99,28 @@ async fn run() -> Result<()> {
 
     let client = build_client(&cfg)?;
 
-    ensure_dataset_exists(&client, &cfg).await?;
-    pull_workspace(&client, &cfg).await?;
+    // ── Startup sync: wait for network, then pull. Non-fatal on failure. ────────
+    // HF Space containers sometimes take 10-30s for DNS/routing to become
+    // available after the container starts. We wait up to 60s, then proceed.
+    eprintln!("waiting for network connectivity to huggingface.co…");
+    match wait_for_network(&client).await {
+        true => {
+            eprintln!("network is up — proceeding with startup sync");
+            if let Err(err) = startup_sync(&client, &cfg).await {
+                eprintln!("startup sync failed (continuing with local workspace): {err:#}");
+            }
+        }
+        false => {
+            eprintln!(
+                "huggingface.co unreachable after {:?} — starting with local workspace",
+                NETWORK_STARTUP_TIMEOUT
+            );
+        }
+    }
+
     rebuild_sync_state(&cfg.workspace).await?;
 
+    // ── Spawn child process ──────────────────────────────────────────────────────
     let mut child = spawn_child_from_args()?;
     let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
     let mut ticker = interval(cfg.sync_interval);
@@ -123,14 +144,14 @@ async fn run() -> Result<()> {
             status = child.wait() => {
                 match status {
                     Ok(s) => {
-                        eprintln!("child exited with status: {s}");
+                        eprintln!("child exited: {s}");
                         if let Err(err) = push_workspace(&client, &cfg).await {
                             eprintln!("final push after child exit failed: {err:#}");
                         }
                         exit(s.code().unwrap_or(1));
                     }
                     Err(err) => {
-                        eprintln!("failed waiting for child: {err:#}");
+                        eprintln!("error waiting for child: {err:#}");
                         break;
                     }
                 }
@@ -141,10 +162,44 @@ async fn run() -> Result<()> {
     Ok(())
 }
 
+/// Poll HF until reachable or timeout. Returns true if network came up.
+async fn wait_for_network(client: &reqwest::Client) -> bool {
+    let probe_url = "https://huggingface.co/api/whoami-v2";
+    let deadline = tokio::time::Instant::now() + NETWORK_STARTUP_TIMEOUT;
+
+    loop {
+        // Use a short per-attempt timeout so we fail fast and retry
+        let attempt = timeout(
+            Duration::from_secs(8),
+            client.get(probe_url).send(),
+        )
+        .await;
+
+        match attempt {
+            // Any HTTP response (even 401) means DNS + routing is working
+            Ok(Ok(_)) => return true,
+            Ok(Err(err)) => eprintln!("network probe error: {err}"),
+            Err(_) => eprintln!("network probe timed out"),
+        }
+
+        if tokio::time::Instant::now() >= deadline {
+            return false;
+        }
+
+        eprintln!("retrying in {:?}…", NETWORK_RETRY_INTERVAL);
+        sleep(NETWORK_RETRY_INTERVAL).await;
+    }
+}
+
+/// Pull workspace from HF dataset on first boot.
+async fn startup_sync(client: &reqwest::Client, cfg: &Config) -> Result<()> {
+    ensure_dataset_exists(client, cfg).await?;
+    pull_workspace(client, cfg).await
+}
+
 fn load_config() -> Result<Config> {
     let token = env::var("HF_TOKEN").context("HF_TOKEN is required")?;
 
-    // Accept OPENCLAW_DATASET_REPO (canonical) with HF_DATASET_ID as fallback.
     let dataset_id = env::var("OPENCLAW_DATASET_REPO")
         .or_else(|_| env::var("HF_DATASET_ID"))
         .context("OPENCLAW_DATASET_REPO (or HF_DATASET_ID) is required")?;
@@ -154,10 +209,10 @@ fn load_config() -> Result<Config> {
     let sync_interval_secs: u64 = env::var("SYNC_INTERVAL")
         .unwrap_or_else(|_| "60".to_string())
         .parse()
-        .context("SYNC_INTERVAL must be an integer in seconds")?;
+        .context("SYNC_INTERVAL must be an integer")?;
 
     if sync_interval_secs == 0 {
-        return Err(anyhow!("SYNC_INTERVAL must be greater than 0"));
+        return Err(anyhow!("SYNC_INTERVAL must be > 0"));
     }
 
     Ok(Config {
@@ -168,14 +223,12 @@ fn load_config() -> Result<Config> {
     })
 }
 
-fn validate_dataset_id(dataset_id: &str) -> Result<()> {
-    let mut parts = dataset_id.split('/');
+fn validate_dataset_id(id: &str) -> Result<()> {
+    let mut parts = id.split('/');
     let owner = parts.next().unwrap_or_default();
     let repo = parts.next().unwrap_or_default();
     if owner.is_empty() || repo.is_empty() || parts.next().is_some() {
-        return Err(anyhow!(
-            "OPENCLAW_DATASET_REPO must be in the form owner/name"
-        ));
+        return Err(anyhow!("OPENCLAW_DATASET_REPO must be owner/name"));
     }
     Ok(())
 }
@@ -186,33 +239,33 @@ fn build_client(cfg: &Config) -> Result<reqwest::Client> {
         AUTHORIZATION,
         format!("Bearer {}", cfg.token)
             .parse()
-            .context("invalid HF_TOKEN for auth header")?,
+            .context("invalid HF_TOKEN")?,
     );
     headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
 
     reqwest::Client::builder()
         .default_headers(headers)
+        .timeout(Duration::from_secs(30))
         .build()
-        .context("failed to create HTTP client")
+        .context("failed to build HTTP client")
 }
 
 async fn ensure_dataset_exists(client: &reqwest::Client, cfg: &Config) -> Result<()> {
     let url = format!("https://huggingface.co/api/datasets/{}", cfg.dataset_id);
-    let response = client.get(&url).send().await?;
+    let resp = client.get(&url).send().await?;
 
-    match response.status() {
+    match resp.status() {
         StatusCode::OK => {
-            let repo: RepoLookup = response.json().await.unwrap_or(RepoLookup { id: None });
-            eprintln!("dataset exists: {}", repo.id.unwrap_or(cfg.dataset_id.clone()));
+            let repo: RepoLookup = resp.json().await.unwrap_or(RepoLookup { id: None });
+            eprintln!("dataset ok: {}", repo.id.unwrap_or(cfg.dataset_id.clone()));
             Ok(())
         }
         StatusCode::NOT_FOUND => {
-            let auto_create = env::var("AUTO_CREATE_DATASET")
+            let auto = env::var("AUTO_CREATE_DATASET")
                 .map(|v| matches!(v.to_lowercase().as_str(), "1" | "true" | "yes" | "on"))
                 .unwrap_or(false);
-
-            if auto_create {
-                eprintln!("dataset {} not found — AUTO_CREATE_DATASET=true, creating…", cfg.dataset_id);
+            if auto {
+                eprintln!("dataset not found — AUTO_CREATE_DATASET=true, creating…");
                 create_private_dataset(client, cfg).await
             } else {
                 Err(anyhow!(
@@ -222,18 +275,15 @@ async fn ensure_dataset_exists(client: &reqwest::Client, cfg: &Config) -> Result
                 ))
             }
         }
-        status => {
-            let body = response.text().await.unwrap_or_default();
-            Err(anyhow!("dataset lookup failed ({status}): {body}"))
-        }
+        s => Err(anyhow!(
+            "dataset lookup failed ({s}): {}",
+            resp.text().await.unwrap_or_default()
+        )),
     }
 }
 
 async fn create_private_dataset(client: &reqwest::Client, cfg: &Config) -> Result<()> {
-    let (owner, name) = cfg
-        .dataset_id
-        .split_once('/')
-        .ok_or_else(|| anyhow!("OPENCLAW_DATASET_REPO must be in the form owner/name"))?;
+    let (owner, name) = cfg.dataset_id.split_once('/').unwrap();
 
     let username = client
         .get("https://huggingface.co/api/whoami-v2")
@@ -254,40 +304,42 @@ async fn create_private_dataset(client: &reqwest::Client, cfg: &Config) -> Resul
         repo_type: "dataset".to_string(),
     };
 
-    let response = client
+    let resp = client
         .post("https://huggingface.co/api/repos/create")
         .json(&req)
         .send()
         .await?;
 
-    if response.status().is_success() || response.status() == StatusCode::CONFLICT {
+    if resp.status().is_success() || resp.status() == StatusCode::CONFLICT {
         Ok(())
     } else {
-        let status = response.status();
-        let body = response.text().await.unwrap_or_default();
-        Err(anyhow!("failed to create dataset ({status}): {body}"))
+        Err(anyhow!(
+            "create dataset failed ({}): {}",
+            resp.status(),
+            resp.text().await.unwrap_or_default()
+        ))
     }
 }
 
 async fn pull_workspace(client: &reqwest::Client, cfg: &Config) -> Result<()> {
-    eprintln!("pulling workspace from HuggingFace: {}", cfg.dataset_id);
-    let remote_files = list_remote_files(client, cfg).await?;
+    eprintln!("pulling workspace from {}", cfg.dataset_id);
+    let files = list_remote_files(client, cfg).await?;
 
-    for file in &remote_files {
+    for file in &files {
         let url = format!(
             "https://huggingface.co/datasets/{}/resolve/main/{}",
             cfg.dataset_id, file
         );
         let bytes = client.get(url).send().await?.error_for_status()?.bytes().await?;
         let target = cfg.workspace.join(file);
-        if let Some(parent) = target.parent() {
-            tokio::fs::create_dir_all(parent).await?;
+        if let Some(p) = target.parent() {
+            tokio::fs::create_dir_all(p).await?;
         }
         let mut f = tokio::fs::File::create(&target).await?;
         f.write_all(&bytes).await?;
     }
 
-    eprintln!("pull complete: {} files restored", remote_files.len());
+    eprintln!("pull complete: {} files", files.len());
     Ok(())
 }
 
@@ -296,16 +348,15 @@ async fn list_remote_files(client: &reqwest::Client, cfg: &Config) -> Result<Vec
         "https://huggingface.co/api/datasets/{}/tree/main?recursive=true",
         cfg.dataset_id
     );
-    let response = client.get(url).send().await?;
-    if response.status() == StatusCode::NOT_FOUND {
-        return Ok(Vec::new());
+    let resp = client.get(url).send().await?;
+    if resp.status() == StatusCode::NOT_FOUND {
+        return Ok(vec![]);
     }
-    let entries: Vec<TreeEntry> = response.error_for_status()?.json().await?;
+    let entries: Vec<TreeEntry> = resp.error_for_status()?.json().await?;
     Ok(entries.into_iter().filter(|e| e.kind == "file").map(|e| e.path).collect())
 }
 
 async fn push_workspace(client: &reqwest::Client, cfg: &Config) -> Result<()> {
-    eprintln!("pushing workspace to HuggingFace…");
     let state_path = cfg.workspace.join(STATE_FILE);
     let mut state = load_state(&state_path).await?;
 
@@ -317,46 +368,43 @@ async fn push_workspace(client: &reqwest::Client, cfg: &Config) -> Result<()> {
         .filter_map(|e| e.ok())
         .filter(|e| e.file_type().is_file())
     {
-        let full_path = entry.path();
-        if full_path == state_path {
+        let full = entry.path();
+        if full == state_path {
             continue;
         }
-
-        let relative = full_path
-            .strip_prefix(&cfg.workspace)
-            .context("failed to strip workspace prefix")?
+        let rel = full
+            .strip_prefix(&cfg.workspace)?
             .to_string_lossy()
             .replace('\\', "/");
 
-        let bytes = tokio::fs::read(full_path).await?;
+        let bytes = tokio::fs::read(full).await?;
         let md5 = format!("{:x}", md5::compute(&bytes));
         let size = bytes.len() as u64;
 
         let changed = state
             .files
-            .get(&relative)
+            .get(&rel)
             .map(|s| s.md5 != md5 || s.size != size)
             .unwrap_or(true);
 
         if changed {
             operations.push(CommitOperation::AddOrUpdate {
-                path: relative.clone(),
+                path: rel.clone(),
                 encoding: "base64".to_string(),
                 content: base64::engine::general_purpose::STANDARD.encode(&bytes),
             });
         }
-
-        current.insert(relative, FileState { md5, size });
+        current.insert(rel, FileState { md5, size });
     }
 
-    let old_paths: HashSet<_> = state.files.keys().cloned().collect();
-    let new_paths: HashSet<_> = current.keys().cloned().collect();
-    for removed in old_paths.difference(&new_paths) {
+    let old: HashSet<_> = state.files.keys().cloned().collect();
+    let new: HashSet<_> = current.keys().cloned().collect();
+    for removed in old.difference(&new) {
         operations.push(CommitOperation::Delete { path: removed.clone() });
     }
 
     if operations.is_empty() {
-        eprintln!("no workspace changes detected");
+        eprintln!("no workspace changes");
         return Ok(());
     }
 
@@ -369,11 +417,13 @@ async fn push_workspace(client: &reqwest::Client, cfg: &Config) -> Result<()> {
         "https://huggingface.co/api/datasets/{}/commit/main",
         cfg.dataset_id
     );
-    let response = client.post(url).json(&req).send().await?;
-    if !response.status().is_success() {
-        let status = response.status();
-        let body = response.text().await.unwrap_or_default();
-        return Err(anyhow!("push failed ({status}): {body}"));
+    let resp = client.post(url).json(&req).send().await?;
+    if !resp.status().is_success() {
+        return Err(anyhow!(
+            "push failed ({}): {}",
+            resp.status(),
+            resp.text().await.unwrap_or_default()
+        ));
     }
 
     state.files = current;
@@ -383,25 +433,24 @@ async fn push_workspace(client: &reqwest::Client, cfg: &Config) -> Result<()> {
 }
 
 async fn rebuild_sync_state(workspace: &Path) -> Result<()> {
-    let mut files = HashMap::<String, FileState>::new();
     let state_path = workspace.join(STATE_FILE);
+    let mut files = HashMap::new();
 
     for entry in WalkDir::new(workspace)
         .into_iter()
         .filter_map(|e| e.ok())
         .filter(|e| e.file_type().is_file())
     {
-        let full_path = entry.path();
-        if full_path == state_path {
+        let full = entry.path();
+        if full == state_path {
             continue;
         }
-        let relative = full_path
-            .strip_prefix(workspace)
-            .context("failed to strip workspace prefix")?
+        let rel = full
+            .strip_prefix(workspace)?
             .to_string_lossy()
             .replace('\\', "/");
-        let bytes = tokio::fs::read(full_path).await?;
-        files.insert(relative, FileState {
+        let bytes = tokio::fs::read(full).await?;
+        files.insert(rel, FileState {
             md5: format!("{:x}", md5::compute(&bytes)),
             size: bytes.len() as u64,
         });
@@ -415,7 +464,7 @@ async fn load_state(path: &Path) -> Result<SyncState> {
         return Ok(SyncState::default());
     }
     let raw = tokio::fs::read(path).await?;
-    Ok(serde_json::from_slice(&raw).context("failed to parse sync state file")?)
+    Ok(serde_json::from_slice(&raw).context("bad sync state")?)
 }
 
 async fn save_state(path: &Path, state: &SyncState) -> Result<()> {
@@ -426,9 +475,7 @@ async fn save_state(path: &Path, state: &SyncState) -> Result<()> {
 fn spawn_child_from_args() -> Result<Child> {
     let args: Vec<String> = env::args().skip(1).collect();
     if args.is_empty() {
-        return Err(anyhow!(
-            "no command provided — pass the child command as entrypoint arguments"
-        ));
+        return Err(anyhow!("no child command provided"));
     }
     Command::new(&args[0])
         .args(&args[1..])
@@ -441,10 +488,10 @@ fn spawn_child_from_args() -> Result<Child> {
 
 fn forward_sigterm(child: &mut Child) {
     if let Some(id) = child.id() {
-        // SAFETY: id is a valid pid from a live child process.
+        // SAFETY: valid pid from live child
         let ret = unsafe { libc::kill(id as libc::pid_t, libc::SIGTERM) };
         if ret != 0 {
-            eprintln!("failed to forward SIGTERM: {}", std::io::Error::last_os_error());
+            eprintln!("SIGTERM forward failed: {}", std::io::Error::last_os_error());
         }
     }
 }
@@ -452,15 +499,12 @@ fn forward_sigterm(child: &mut Child) {
 async fn wait_for_child_shutdown(child: &mut Child) {
     match timeout(FINAL_WAIT_TIMEOUT, child.wait()).await {
         Ok(Ok(s)) => eprintln!("child exited after SIGTERM: {s}"),
-        Ok(Err(e)) => eprintln!("error waiting for child after SIGTERM: {e:#}"),
+        Ok(Err(e)) => eprintln!("wait error: {e:#}"),
         Err(_) => {
-            eprintln!("child did not exit within {:?} — sending SIGKILL", FINAL_WAIT_TIMEOUT);
+            eprintln!("child timeout — sending SIGKILL");
             if let Some(id) = child.id() {
-                // SAFETY: id is a valid pid from a live child process.
-                let ret = unsafe { libc::kill(id as libc::pid_t, libc::SIGKILL) };
-                if ret != 0 {
-                    eprintln!("failed to SIGKILL: {}", std::io::Error::last_os_error());
-                }
+                // SAFETY: valid pid from live child
+                unsafe { libc::kill(id as libc::pid_t, libc::SIGKILL) };
             }
         }
     }