feat: jobs survive restarts across the network (worker drain + gateway persistence); v0.2.92

Cargo.toml

@@ -23,7 +23,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.2.91"
+version = "0.2.92"
 edition = "2024"
 license = "MIT"
 repository = "https://github.com/profullstack/c0mpute"

apps/tui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@c0mpute/tui",
-  "version": "0.2.91",
+  "version": "0.2.92",
   "private": true,
   "type": "module",
   "bin": {

apps/web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "c0mpute",
-  "version": "0.2.91",
+  "version": "0.2.92",
   "private": true,
   "type": "module",
   "scripts": {

apps/web/src/app/releases/latest.json/route.ts

@@ -28,7 +28,7 @@ export const dynamic = "force-dynamic";
  * this in sync with the workspace version in `Cargo.toml` on each release so a
  * freshly built CLI reports "already latest" rather than a phantom upgrade.
  */
-const FALLBACK_VERSION = "0.2.91";
+const FALLBACK_VERSION = "0.2.92";
 const FALLBACK_MIN_REQUIRED = "0.0.1";
 
 type Channel = "stable" | "beta" | "nightly";

node/crates/c0mpute-core/src/diffusion.rs

@@ -654,6 +654,8 @@ pub fn spawn_diffusion_runner(
             };
             let started = Instant::now();
             let job_id = job.accept.job_id.clone();
+            // An auto-update waits for this to finish instead of killing it.
+            let _long = c0mpute_update::LongJob::start();
             let result = run_job(&net, &store, &cache_root, &job).await;
             cleanup_job_dir(&cache_root, &job_id).await;
             match result {

node/crates/c0mpute-core/src/job_progress.rs

@@ -111,11 +111,89 @@ pub enum Stored {
 const MAX_RESULTS: usize = 48;
 
 #[derive(Clone, Default)]
-pub struct ResultStore(Arc<Mutex<HashMap<String, (u64, Stored)>>>);
+pub struct ResultStore(Arc<Mutex<HashMap<String, (u64, Stored)>>>, Option<Arc<std::path::PathBuf>>);
+
+/// On-disk form of a stored result's metadata (bytes live beside it).
+#[derive(serde::Serialize, serde::Deserialize)]
+struct StoredMeta {
+    at_ms: u64,
+    ok: bool,
+    content_type: String,
+    job_id: String,
+    worker: String,
+    status: u16,
+    message: String,
+}
+
+fn static_content_type(t: &str) -> &'static str {
+    match t {
+        "audio/flac" => "audio/flac",
+        "audio/wav" => "audio/wav",
+        "image/png" => "image/png",
+        "video/mp4" => "video/mp4",
+        _ => "audio/mpeg",
+    }
+}
 
 impl ResultStore {
+    /// A store that also keeps results under `dir`, so they outlive a restart.
+    pub fn persistent(dir: std::path::PathBuf) -> Self {
+        let _ = std::fs::create_dir_all(&dir);
+        // Drop files older than KEEP_MS left by earlier runs.
+        let now = now_ms();
+        if let Ok(entries) = std::fs::read_dir(&dir) {
+            for e in entries.flatten() {
+                let old = e
+                    .metadata()
+                    .and_then(|m| m.modified())
+                    .ok()
+                    .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
+                    .is_some_and(|d| now.saturating_sub(d.as_millis() as u64) > KEEP_MS);
+                if old {
+                    let _ = std::fs::remove_file(e.path());
+                }
+            }
+        }
+        Self(Arc::default(), Some(Arc::new(dir)))
+    }
+
+    fn write_disk(&self, key: &str, at_ms: u64, stored: &Stored) {
+        let Some(dir) = self.1.as_deref() else { return };
+        let meta = match stored {
+            Stored::Done { bytes, content_type, job_id, worker } => {
+                if std::fs::write(dir.join(format!("{key}.bin")), bytes.as_slice()).is_err() {
+                    return;
+                }
+                StoredMeta { at_ms, ok: true, content_type: content_type.to_string(), job_id: job_id.clone(), worker: worker.clone(), status: 200, message: String::new() }
+            }
+            Stored::Failed { status, message } => StoredMeta {
+                at_ms, ok: false, content_type: String::new(), job_id: String::new(), worker: String::new(), status: *status, message: message.clone(),
+            },
+        };
+        if let Ok(json) = serde_json::to_vec(&meta) {
+            let _ = std::fs::write(dir.join(format!("{key}.json")), json);
+        }
+    }
+
+    fn read_disk(&self, key: &str) -> Option<(u64, Stored)> {
+        let dir = self.1.as_deref()?;
+        let meta: StoredMeta = serde_json::from_slice(&std::fs::read(dir.join(format!("{key}.json"))).ok()?).ok()?;
+        let stored = if meta.ok {
+            Stored::Done {
+                bytes: Arc::new(std::fs::read(dir.join(format!("{key}.bin"))).ok()?),
+                content_type: static_content_type(&meta.content_type),
+                job_id: meta.job_id,
+                worker: meta.worker,
+            }
+        } else {
+            Stored::Failed { status: meta.status, message: meta.message }
+        };
+        Some((meta.at_ms, stored))
+    }
+
     pub fn put(&self, key: &str, stored: Stored) {
         let now = now_ms();
+        self.write_disk(key, now, &stored);
         let Ok(mut map) = self.0.lock() else { return };
         map.retain(|_, (at, _)| now.saturating_sub(*at) < KEEP_MS);
         while map.len() >= MAX_RESULTS {
@@ -129,11 +207,72 @@ impl ResultStore {
 
     pub fn get(&self, key: &str) -> Option<Stored> {
         let now = now_ms();
-        let map = self.0.lock().ok()?;
-        map.get(key)
-            .filter(|(at, _)| now.saturating_sub(*at) < KEEP_MS)
-            .map(|(_, s)| s.clone())
+        let hit = {
+            let map = self.0.lock().ok()?;
+            map.get(key).cloned()
+        };
+        // After a restart the memory is empty; the disk copy still answers.
+        let (at, stored) = hit.or_else(|| self.read_disk(key))?;
+        (now.saturating_sub(at) < KEEP_MS).then_some(stored)
+    }
+}
+
+/// A tracked request that has not finished: enough to run it again after a
+/// gateway restart, under the same request id.
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct PendingJob {
+    pub workload: String,
+    pub spec: serde_json::Value,
+    pub required: Vec<String>,
+    pub max_bytes: usize,
+    pub started_at_ms: u64,
+}
+
+/// Unfinished tracked requests on disk (`<dir>/<id>.json`).
+#[derive(Clone, Default)]
+pub struct PendingStore(Option<Arc<std::path::PathBuf>>);
+
+impl PendingStore {
+    pub fn persistent(dir: std::path::PathBuf) -> Self {
+        let _ = std::fs::create_dir_all(&dir);
+        Self(Some(Arc::new(dir)))
     }
+
+    pub fn save(&self, key: &str, job: &PendingJob) {
+        let Some(dir) = self.0.as_deref() else { return };
+        if let Ok(json) = serde_json::to_vec(job) {
+            let _ = std::fs::write(dir.join(format!("{key}.json")), json);
+        }
+    }
+
+    pub fn remove(&self, key: &str) {
+        if let Some(dir) = self.0.as_deref() {
+            let _ = std::fs::remove_file(dir.join(format!("{key}.json")));
+        }
+    }
+
+    /// Unfinished requests younger than the keep window; older ones are dropped.
+    pub fn take_resumable(&self) -> Vec<(String, PendingJob)> {
+        let Some(dir) = self.0.as_deref() else { return Vec::new() };
+        let now = now_ms();
+        let mut out = Vec::new();
+        for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
+            let path = e.path();
+            let Some(key) = path.file_stem().and_then(|s| s.to_str()).map(str::to_string) else { continue };
+            let job = std::fs::read(&path).ok().and_then(|b| serde_json::from_slice::<PendingJob>(&b).ok());
+            match job {
+                Some(job) if valid_key(&key) && now.saturating_sub(job.started_at_ms) < KEEP_MS => out.push((key, job)),
+                _ => {
+                    let _ = std::fs::remove_file(&path);
+                }
+            }
+        }
+        out
+    }
+}
+
+pub fn now_ms_pub() -> u64 {
+    now_ms()
 }
 
 /// A caller-chosen request id: short and URL-safe.
@@ -212,6 +351,38 @@ mod tests {
         assert!(store.0.lock().unwrap().len() <= MAX_RESULTS, "bounded");
     }
 
+    #[test]
+    fn results_and_pending_jobs_survive_a_restart_on_disk() {
+        let dir = std::env::temp_dir().join(format!("c0-jobs-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        {
+            let results = ResultStore::persistent(dir.join("results"));
+            results.put(
+                "song-9",
+                Stored::Done { bytes: Arc::new(b"ID3song".to_vec()), content_type: "audio/mpeg", job_id: "j9".into(), worker: "w9".into() },
+            );
+            let pending = PendingStore::persistent(dir.join("jobs"));
+            pending.save(
+                "song-10",
+                &PendingJob { workload: "music.txt2music".into(), spec: serde_json::json!({"prompt": "x"}), required: vec![], max_bytes: 1, started_at_ms: now_ms() },
+            );
+            pending.save(
+                "stale",
+                &PendingJob { workload: "music.txt2music".into(), spec: serde_json::json!({}), required: vec![], max_bytes: 1, started_at_ms: 1 },
+            );
+        }
+        // "Restart": fresh stores over the same directories.
+        let results = ResultStore::persistent(dir.join("results"));
+        assert!(matches!(results.get("song-9"), Some(Stored::Done { ref bytes, content_type: "audio/mpeg", ref worker, .. }) if bytes.as_slice() == b"ID3song" && worker == "w9"));
+        let pending = PendingStore::persistent(dir.join("jobs"));
+        let resumable = pending.take_resumable();
+        assert_eq!(resumable.len(), 1, "the stale one is dropped");
+        assert_eq!(resumable[0].0, "song-10");
+        pending.remove("song-10");
+        assert!(pending.take_resumable().is_empty());
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     fn request_ids_must_be_short_and_url_safe() {
         assert!(valid_key("17a1b21d-a95e-4345-a3fe-d14dfd6868e3"));