feat: activate/deactivate/list/sync/pull commands + peer sync queue

- activate <org/repo>: wire hot tier, write .cargo/config.toml, register
- deactivate <org/repo>: auto-sync to peer, evict hot tier, unregister
- list: show all warm repos with activation status and last-active time
- sync <org/repo>: git push origin then SSH peer to pull
- pull <org/repo>: git pull from origin, stamp last_sync in manifest
- SyncQueue: file-based retry queue (/var/lib/claw-store/sync-queue.toml)
- daemon: drains sync queue on every 5-min poll tick
- config: [peer] host/user for cross-node notification
- manifest: Project.name is now org/repo, adds last_sync field
- hot tier paths are org/repo-namespaced to avoid collisions

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
Omar Sobh
2026-06-16 11:45:04 +00:00
co-authored by Claude Sonnet 4.6
parent 3fb108edfc
commit a9c12173fe
6 changed files with 423 additions and 65 deletions
+7
View File
@@ -45,6 +45,12 @@ pub struct ReplicationConfig {
pub nightly_at: Option<String>, pub nightly_at: Option<String>,
} }
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct PeerConfig {
pub host: String,
pub user: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)] #[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Config { pub struct Config {
pub node: NodeConfig, pub node: NodeConfig,
@@ -52,6 +58,7 @@ pub struct Config {
pub warm: WarmConfig, pub warm: WarmConfig,
pub cold: Option<ColdConfig>, pub cold: Option<ColdConfig>,
pub replication: Option<ReplicationConfig>, pub replication: Option<ReplicationConfig>,
pub peer: Option<PeerConfig>,
} }
impl Config { impl Config {
+12
View File
@@ -2,6 +2,7 @@ use crate::config::Config;
use crate::hot; use crate::hot;
use crate::manifest::Manifest; use crate::manifest::Manifest;
use crate::snapshot; use crate::snapshot;
use crate::sync::{SyncQueue, drain_sync_queue};
use crate::zfs::SystemZfs; use crate::zfs::SystemZfs;
use anyhow::Result; use anyhow::Result;
use sysinfo::{ProcessRefreshKind, RefreshKind, System}; use sysinfo::{ProcessRefreshKind, RefreshKind, System};
@@ -28,6 +29,17 @@ pub async fn run(cfg: Config, mut manifest: Manifest) -> Result<()> {
hot::gc_by_space(&mut manifest, cfg.hot.max_gb as f64)?; hot::gc_by_space(&mut manifest, cfg.hot.max_gb as f64)?;
manifest.save(&manifest_path)?; manifest.save(&manifest_path)?;
} }
// Retry any pending sync notifications
if let Some(peer) = &cfg.peer {
let queue_path = SyncQueue::default_path();
let mut queue = SyncQueue::load(&queue_path).unwrap_or_default();
if !queue.jobs.is_empty() {
tracing::info!("retrying {} queued sync job(s)", queue.jobs.len());
if let Err(e) = drain_sync_queue(&mut queue, &peer.user, &peer.host, &queue_path) {
tracing::error!("sync queue drain failed: {:#}", e);
}
}
}
} }
_ = snap_tick.tick() => { _ = snap_tick.tick() => {
let ts = chrono::Utc::now().format("%Y-%m-%d-%H%M").to_string(); let ts = chrono::Utc::now().format("%Y-%m-%d-%H%M").to_string();
+1
View File
@@ -83,6 +83,7 @@ mod tests {
hot_target_path: path.clone(), hot_target_path: path.clone(),
last_build: None, last_build: None,
last_active: None, last_active: None,
last_sync: None,
} }
} }
+219 -60
View File
@@ -5,9 +5,10 @@ mod hot;
mod manifest; mod manifest;
mod restore; mod restore;
mod snapshot; mod snapshot;
mod sync;
mod zfs; mod zfs;
use anyhow::{Context, Result}; use anyhow::{bail, Context, Result};
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
use config::Config; use config::Config;
use manifest::{Manifest, Project}; use manifest::{Manifest, Project};
@@ -25,13 +26,21 @@ struct Cli {
#[derive(Subcommand)] #[derive(Subcommand)]
enum Cmd { enum Cmd {
/// Register a project: write .cargo/config.toml pointing to hot tier /// Register a project and wire up hot tier (format: org/repo)
Init { Activate {
name: String, project: String,
#[arg(long)] #[arg(long, help = "Clone from this URL if not present in warm tier")]
repo: Option<String>, clone: Option<String>,
}, },
/// Run the background daemon (snapshot cron, GC, replication) /// Sync to peer then remove from hot tier (keeps warm clone)
Deactivate { project: String },
/// List all repos in warm tier with activation status
List,
/// Push commits to origin then notify peer to pull
Sync { project: String },
/// Pull latest from origin into warm tier (also invoked by peer)
Pull { project: String },
/// Run the background daemon (snapshot cron, GC, replication, sync retry)
Daemon, Daemon,
/// Show tier usage, active projects, recent snapshots /// Show tier usage, active projects, recent snapshots
Status, Status,
@@ -39,7 +48,7 @@ enum Cmd {
Gc, Gc,
/// Take a ZFS snapshot of the warm tier now /// Take a ZFS snapshot of the warm tier now
Snapshot, Snapshot,
/// List available snapshots for a project /// List available snapshots
ListSnapshots { project: String }, ListSnapshots { project: String },
/// Restore a project from a snapshot /// Restore a project from a snapshot
Restore { project: String, snapshot: String }, Restore { project: String, snapshot: String },
@@ -61,7 +70,13 @@ async fn main() -> Result<()> {
let zfs = SystemZfs; let zfs = SystemZfs;
match cli.cmd { match cli.cmd {
Cmd::Init { name, repo } => cmd_init(&cfg, &mut manifest, &manifest_path, &name, repo.as_deref())?, Cmd::Activate { project, clone } =>
cmd_activate(&cfg, &mut manifest, &manifest_path, &project, clone.as_deref())?,
Cmd::Deactivate { project } =>
cmd_deactivate(&cfg, &mut manifest, &manifest_path, &project)?,
Cmd::List => cmd_list(&cfg, &manifest)?,
Cmd::Sync { project } => cmd_sync(&cfg, &manifest, &project)?,
Cmd::Pull { project } => cmd_pull(&cfg, &mut manifest, &manifest_path, &project)?,
Cmd::Daemon => daemon::run(cfg, manifest).await?, Cmd::Daemon => daemon::run(cfg, manifest).await?,
Cmd::Status => cmd_status(&cfg, &manifest, &zfs)?, Cmd::Status => cmd_status(&cfg, &manifest, &zfs)?,
Cmd::Gc => cmd_gc(&cfg, &mut manifest)?, Cmd::Gc => cmd_gc(&cfg, &mut manifest)?,
@@ -73,45 +88,196 @@ async fn main() -> Result<()> {
Ok(()) Ok(())
} }
fn cmd_init( // ── activate ─────────────────────────────────────────────────────────────────
fn cmd_activate(
cfg: &Config, cfg: &Config,
manifest: &mut Manifest, manifest: &mut Manifest,
manifest_path: &std::path::Path, manifest_path: &std::path::Path,
name: &str, project: &str,
repo: Option<&str>, clone_url: Option<&str>,
) -> Result<()> { ) -> Result<()> {
let warm = cfg.warm.projects_path.join(name); let warm = warm_path(cfg, project);
let hot_target = cfg.hot.path.join(name); let hot_target = hot_path(cfg, project);
if let Some(url) = repo {
if !warm.exists() { if !warm.exists() {
match clone_url {
Some(url) => {
println!("Cloning {} → {}", url, warm.display()); println!("Cloning {} → {}", url, warm.display());
let out = std::process::Command::new("git") let status = std::process::Command::new("git")
.args(["clone", url, warm.to_str().unwrap()]) .args(["clone", url, warm.to_str().unwrap()])
.status()?; .status()?;
anyhow::ensure!(out.success(), "git clone failed"); anyhow::ensure!(status.success(), "git clone failed");
}
None => bail!(
"warm path does not exist: {}\nUse --clone <url> to clone it first.",
warm.display()
),
} }
} else {
std::fs::create_dir_all(&warm)?;
} }
std::fs::create_dir_all(&hot_target)?; std::fs::create_dir_all(&hot_target)?;
cargo_init::write_cargo_config(&warm, &hot_target)?; cargo_init::write_cargo_config(&warm, &hot_target)?;
manifest.upsert(Project { manifest.upsert(Project {
name: name.into(), name: project.to_string(),
warm_path: warm.clone(), warm_path: warm.clone(),
hot_target_path: hot_target, hot_target_path: hot_target.clone(),
last_build: None, last_build: None,
last_active: None, last_active: None,
last_sync: None,
}); });
manifest.save(manifest_path)?; manifest.save(manifest_path)?;
println!("registered: {}", name);
println!("activated: {}", project);
println!(" source → {}", warm.display()); println!(" source → {}", warm.display());
println!(" target/ → {}/", cfg.hot.path.join(name).display()); println!(" target/ → {}", hot_target.display());
Ok(()) Ok(())
} }
// ── deactivate ────────────────────────────────────────────────────────────────
fn cmd_deactivate(
cfg: &Config,
manifest: &mut Manifest,
manifest_path: &std::path::Path,
project: &str,
) -> Result<()> {
// Sync first if peer is configured
if let Some(peer) = &cfg.peer {
println!("Syncing {} to peer before deactivate...", project);
if let Err(e) = cmd_sync(cfg, manifest, project) {
eprintln!(" warning: sync failed ({}), queuing for retry", e);
let p = manifest.get(project);
let warm = p.map(|p| p.warm_path.clone()).unwrap_or_else(|| warm_path(cfg, project));
let mut queue = sync::SyncQueue::load(&sync::SyncQueue::default_path()).unwrap_or_default();
queue.enqueue(project, &warm);
queue.save(&sync::SyncQueue::default_path())?;
}
}
// Evict hot tier
let hot_target = hot_path(cfg, project);
if hot_target.exists() {
std::fs::remove_dir_all(&hot_target)?;
println!("evicted hot target: {}", hot_target.display());
}
// Remove cargo config
let warm = warm_path(cfg, project);
let cargo_config = warm.join(".cargo/config.toml");
if cargo_config.exists() {
std::fs::remove_file(&cargo_config)?;
}
manifest.projects.retain(|p| p.name != project);
manifest.save(manifest_path)?;
println!("deactivated: {} (warm clone kept at {})", project, warm.display());
Ok(())
}
// ── list ──────────────────────────────────────────────────────────────────────
fn cmd_list(cfg: &Config, manifest: &Manifest) -> Result<()> {
let base = &cfg.warm.projects_path;
println!("{:<45} {:<10} {}", "PROJECT", "STATUS", "LAST ACTIVE");
println!("{}", "-".repeat(75));
let mut rows: Vec<(String, bool, String)> = Vec::new();
// Walk org/repo structure
if let Ok(orgs) = std::fs::read_dir(base) {
for org_entry in orgs.flatten() {
let org_path = org_entry.path();
if !org_path.is_dir() { continue; }
let org = org_entry.file_name().to_string_lossy().to_string();
if let Ok(repos) = std::fs::read_dir(&org_path) {
for repo_entry in repos.flatten() {
let repo_path = repo_entry.path();
if !repo_path.is_dir() { continue; }
let repo = repo_entry.file_name().to_string_lossy().to_string();
let key = format!("{}/{}", org, repo);
let active = manifest.get(&key);
let last = active
.and_then(|p| p.last_active)
.map(|t| t.format("%Y-%m-%d %H:%M").to_string())
.unwrap_or_else(|| "—".into());
rows.push((key, active.is_some(), last));
}
}
}
}
// Also include top-level repos (legacy flat layout)
if let Ok(entries) = std::fs::read_dir(base) {
for entry in entries.flatten() {
let p = entry.path();
if p.is_dir() && p.join(".git").exists() {
let name = entry.file_name().to_string_lossy().to_string();
let active = manifest.get(&name);
let last = active
.and_then(|p| p.last_active)
.map(|t| t.format("%Y-%m-%d %H:%M").to_string())
.unwrap_or_else(|| "—".into());
rows.push((name, active.is_some(), last));
}
}
}
rows.sort_by(|a, b| a.0.cmp(&b.0));
rows.dedup_by(|a, b| a.0 == b.0);
let active_count = rows.iter().filter(|r| r.1).count();
for (name, is_active, last) in &rows {
let status = if *is_active { "active" } else { "warm" };
println!("{:<45} {:<10} {}", name, status, last);
}
println!("\n{} repos ({} active, {} warm-only)", rows.len(), active_count, rows.len() - active_count);
Ok(())
}
// ── sync ─────────────────────────────────────────────────────────────────────
fn cmd_sync(cfg: &Config, manifest: &Manifest, project: &str) -> Result<()> {
let peer = cfg.peer.as_ref()
.context("no [peer] configured — add peer.host and peer.user to config.toml")?;
let warm = manifest.get(project)
.map(|p| p.warm_path.clone())
.unwrap_or_else(|| warm_path(cfg, project));
println!("Pushing {}...", project);
sync::sync_project(&warm, project, &peer.user, &peer.host)?;
println!(" pushed to origin");
println!(" notified {}@{} to pull", peer.user, peer.host);
Ok(())
}
// ── pull ─────────────────────────────────────────────────────────────────────
fn cmd_pull(
cfg: &Config,
manifest: &mut Manifest,
manifest_path: &std::path::Path,
project: &str,
) -> Result<()> {
let warm = manifest.get(project)
.map(|p| p.warm_path.clone())
.unwrap_or_else(|| warm_path(cfg, project));
sync::pull_project(&warm)?;
// Stamp last_sync on the manifest entry if it exists
if let Some(p) = manifest.get_mut(project) {
p.last_sync = Some(chrono::Utc::now());
manifest.save(manifest_path)?;
}
println!("pulled: {}", project);
Ok(())
}
// ── existing commands ─────────────────────────────────────────────────────────
fn cmd_status(cfg: &Config, manifest: &Manifest, zfs: &SystemZfs) -> Result<()> { fn cmd_status(cfg: &Config, manifest: &Manifest, zfs: &SystemZfs) -> Result<()> {
println!("=== claw-store status — {} ===\n", cfg.node.name); println!("=== claw-store status — {} ===\n", cfg.node.name);
println!("HOT tier: {}", cfg.hot.path.display()); println!("HOT tier: {}", cfg.hot.path.display());
@@ -121,32 +287,30 @@ fn cmd_status(cfg: &Config, manifest: &Manifest, zfs: &SystemZfs) -> Result<()>
let snaps = zfs.list_snapshots(&cfg.warm.zfs_dataset).unwrap_or_default(); let snaps = zfs.list_snapshots(&cfg.warm.zfs_dataset).unwrap_or_default();
println!(" snapshots: {}\n", snaps.len()); println!(" snapshots: {}\n", snaps.len());
if let Some(cold) = &cfg.cold { if let Some(cold) = &cfg.cold {
println!("COLD tier: {}", cold.archive_path.display()); println!("COLD tier: {}\n", cold.archive_path.display());
} }
println!("Projects ({}):", manifest.projects.len()); if let Some(peer) = &cfg.peer {
println!("PEER: {}@{}\n", peer.user, peer.host);
}
println!("Active projects ({}):", manifest.projects.len());
for p in &manifest.projects { for p in &manifest.projects {
let active = p let active = p.last_active
.last_active .map(|t| t.format("%Y-%m-%d %H:%M").to_string())
.map(|t| format!("{}", t.format("%Y-%m-%d %H:%M")))
.unwrap_or_else(|| "never".into()); .unwrap_or_else(|| "never".into());
println!(" {} (last active: {})", p.name, active); let synced = p.last_sync
.map(|t| t.format("%Y-%m-%d %H:%M").to_string())
.unwrap_or_else(|| "never".into());
println!(" {} (active: {}, synced: {})", p.name, active, synced);
} }
Ok(()) Ok(())
} }
fn cmd_gc(cfg: &Config, manifest: &mut Manifest) -> Result<()> { fn cmd_gc(cfg: &Config, manifest: &mut Manifest) -> Result<()> {
let stale = hot::gc_stale_targets(manifest, cfg.hot.stale_hours)?; let stale = hot::gc_stale_targets(manifest, cfg.hot.stale_hours)?;
if stale.is_empty() { if stale.is_empty() { println!("Nothing to evict."); return Ok(()); }
println!("Nothing to evict."); for name in &stale { println!(" evicted (stale): {}", name); }
return Ok(());
}
for name in &stale {
println!(" evicted (stale): {}", name);
}
let space = hot::gc_by_space(manifest, cfg.hot.max_gb as f64)?; let space = hot::gc_by_space(manifest, cfg.hot.max_gb as f64)?;
for name in &space { for name in &space { println!(" evicted (space): {}", name); }
println!(" evicted (space): {}", name);
}
manifest.save(&Manifest::default_path())?; manifest.save(&Manifest::default_path())?;
Ok(()) Ok(())
} }
@@ -154,9 +318,7 @@ fn cmd_gc(cfg: &Config, manifest: &mut Manifest) -> Result<()> {
fn cmd_snapshot(cfg: &Config, zfs: &SystemZfs) -> Result<()> { fn cmd_snapshot(cfg: &Config, zfs: &SystemZfs) -> Result<()> {
let ts = chrono::Utc::now().format("%Y-%m-%d-%H%M").to_string(); let ts = chrono::Utc::now().format("%Y-%m-%d-%H%M").to_string();
snapshot::run_snapshot_cycle( snapshot::run_snapshot_cycle(
zfs, zfs, &cfg.warm.zfs_dataset, &ts,
&cfg.warm.zfs_dataset,
&ts,
cfg.warm.snapshot_retain_hours as usize, cfg.warm.snapshot_retain_hours as usize,
cfg.warm.snapshot_retain_days as usize, cfg.warm.snapshot_retain_days as usize,
cfg.warm.snapshot_retain_weeks as usize, cfg.warm.snapshot_retain_weeks as usize,
@@ -167,13 +329,8 @@ fn cmd_snapshot(cfg: &Config, zfs: &SystemZfs) -> Result<()> {
fn cmd_list_snapshots(cfg: &Config, zfs: &SystemZfs, _project: &str) -> Result<()> { fn cmd_list_snapshots(cfg: &Config, zfs: &SystemZfs, _project: &str) -> Result<()> {
let snaps = zfs.list_snapshots(&cfg.warm.zfs_dataset)?; let snaps = zfs.list_snapshots(&cfg.warm.zfs_dataset)?;
if snaps.is_empty() { if snaps.is_empty() { println!("No snapshots found."); return Ok(()); }
println!("No snapshots found."); for s in &snaps { println!(" {}", s); }
return Ok(());
}
for s in &snaps {
println!(" {}", s);
}
Ok(()) Ok(())
} }
@@ -183,25 +340,27 @@ fn cmd_restore(cfg: &Config, zfs: &SystemZfs, project: &str, snap: &str) -> Resu
anyhow::ensure!(found.is_some(), "snapshot '{}' not found", snap); anyhow::ensure!(found.is_some(), "snapshot '{}' not found", snap);
let path = restore::restore_project(zfs, project, &full_snap, &cfg.warm.zfs_dataset)?; let path = restore::restore_project(zfs, project, &full_snap, &cfg.warm.zfs_dataset)?;
println!("Restored to: {}", path.display()); println!("Restored to: {}", path.display());
println!( println!("Cleanup: zfs destroy {}/{}-restore-{}", cfg.warm.zfs_dataset, project, snap);
"Cleanup: zfs destroy {}/{}-restore-{}",
cfg.warm.zfs_dataset, project, snap
);
Ok(()) Ok(())
} }
fn cmd_replicate(cfg: &Config, zfs: &SystemZfs) -> Result<()> { fn cmd_replicate(cfg: &Config, zfs: &SystemZfs) -> Result<()> {
let rep = cfg let rep = cfg.replication.as_ref()
.replication
.as_ref()
.context("no replication config — this node does not replicate")?; .context("no replication config — this node does not replicate")?;
let host = rep.send_to_host.as_ref().context("send_to_host not set")?; let host = rep.send_to_host.as_ref().context("send_to_host not set")?;
let user = rep.send_to_user.as_ref().context("send_to_user not set")?; let user = rep.send_to_user.as_ref().context("send_to_user not set")?;
let dest = rep let dest = rep.cold_dataset_on_peer.as_ref().context("cold_dataset_on_peer not set")?;
.cold_dataset_on_peer
.as_ref()
.context("cold_dataset_on_peer not set")?;
snapshot::replicate_to_cold(zfs, &cfg.warm.zfs_dataset, user, host, dest)?; snapshot::replicate_to_cold(zfs, &cfg.warm.zfs_dataset, user, host, dest)?;
println!("Replication to {}@{} complete.", user, host); println!("Replication to {}@{} complete.", user, host);
Ok(()) Ok(())
} }
// ── path helpers ──────────────────────────────────────────────────────────────
fn warm_path(cfg: &Config, project: &str) -> PathBuf {
cfg.warm.projects_path.join(project)
}
fn hot_path(cfg: &Config, project: &str) -> PathBuf {
cfg.hot.path.join(project)
}
+4 -1
View File
@@ -5,11 +5,12 @@ use std::path::{Path, PathBuf};
#[derive(Debug, Clone, Deserialize, Serialize)] #[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Project { pub struct Project {
pub name: String, pub name: String, // "org/repo" e.g. "quantumclaw/quantum-pulse"
pub warm_path: PathBuf, pub warm_path: PathBuf,
pub hot_target_path: PathBuf, pub hot_target_path: PathBuf,
pub last_build: Option<DateTime<Utc>>, pub last_build: Option<DateTime<Utc>>,
pub last_active: Option<DateTime<Utc>>, pub last_active: Option<DateTime<Utc>>,
pub last_sync: Option<DateTime<Utc>>,
} }
#[derive(Debug, Clone, Deserialize, Serialize, Default)] #[derive(Debug, Clone, Deserialize, Serialize, Default)]
@@ -68,6 +69,7 @@ mod tests {
hot_target_path: "/hot/targets/kitchen-cash".into(), hot_target_path: "/hot/targets/kitchen-cash".into(),
last_build: None, last_build: None,
last_active: None, last_active: None,
last_sync: None,
}); });
let f = NamedTempFile::new().unwrap(); let f = NamedTempFile::new().unwrap();
m.save(f.path()).unwrap(); m.save(f.path()).unwrap();
@@ -85,6 +87,7 @@ mod tests {
hot_target_path: "/hot/targets/zeroclaw".into(), hot_target_path: "/hot/targets/zeroclaw".into(),
last_build: None, last_build: None,
last_active: None, last_active: None,
last_sync: None,
}); });
assert!(m.get("zeroclaw").is_some()); assert!(m.get("zeroclaw").is_some());
assert!(m.get("nonexistent").is_none()); assert!(m.get("nonexistent").is_none());
+176
View File
@@ -0,0 +1,176 @@
use anyhow::{bail, Context, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct SyncJob {
pub project: String, // "org/repo"
pub warm_path: PathBuf,
pub queued_at: DateTime<Utc>,
pub attempts: u32,
}
#[derive(Debug, Clone, Deserialize, Serialize, Default)]
pub struct SyncQueue {
pub jobs: Vec<SyncJob>,
}
impl SyncQueue {
pub fn load(path: &Path) -> Result<Self> {
if !path.exists() { return Ok(Self::default()); }
let s = std::fs::read_to_string(path)
.with_context(|| format!("reading sync queue at {}", path.display()))?;
toml::from_str(&s).context("parsing sync queue")
}
pub fn save(&self, path: &Path) -> Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(path, toml::to_string_pretty(self)?)
.context("writing sync queue")
}
pub fn enqueue(&mut self, project: &str, warm_path: &Path) {
// Deduplicate — only one pending job per project
self.jobs.retain(|j| j.project != project);
self.jobs.push(SyncJob {
project: project.to_string(),
warm_path: warm_path.to_path_buf(),
queued_at: Utc::now(),
attempts: 0,
});
}
pub fn default_path() -> PathBuf {
PathBuf::from("/var/lib/claw-store/sync-queue.toml")
}
}
/// Push local commits to origin then notify peer to pull.
pub fn sync_project(
warm_path: &Path,
project: &str,
peer_user: &str,
peer_host: &str,
) -> Result<()> {
git_push(warm_path)?;
notify_peer(project, peer_user, peer_host)
}
/// Pull latest from origin into warm_path.
pub fn pull_project(warm_path: &Path) -> Result<()> {
if !warm_path.exists() {
bail!("warm path does not exist: {}", warm_path.display());
}
let out = std::process::Command::new("git")
.args(["-C", warm_path.to_str().unwrap(), "pull", "--ff-only"])
.output()
.context("running git pull")?;
if !out.status.success() {
bail!("git pull failed: {}", String::from_utf8_lossy(&out.stderr));
}
tracing::info!("pulled {}: {}", project_label(warm_path),
String::from_utf8_lossy(&out.stdout).trim());
Ok(())
}
/// Process pending sync queue — retry each job, drop successes, bump attempt counter.
pub fn drain_sync_queue(
queue: &mut SyncQueue,
peer_user: &str,
peer_host: &str,
queue_path: &Path,
) -> Result<()> {
let mut remaining = Vec::new();
for mut job in queue.jobs.drain(..) {
job.attempts += 1;
match notify_peer(&job.project, peer_user, peer_host) {
Ok(()) => tracing::info!("sync queue: notified peer for {}", job.project),
Err(e) => {
tracing::warn!("sync queue: peer notify failed for {} (attempt {}): {:#}",
job.project, job.attempts, e);
if job.attempts < 48 { // drop after 48 attempts (~4h at 5min intervals)
remaining.push(job);
}
}
}
}
queue.jobs = remaining;
queue.save(queue_path)
}
fn git_push(warm_path: &Path) -> Result<()> {
let out = std::process::Command::new("git")
.args(["-C", warm_path.to_str().unwrap(), "push"])
.output()
.context("running git push")?;
if !out.status.success() {
bail!("git push failed: {}", String::from_utf8_lossy(&out.stderr));
}
Ok(())
}
fn notify_peer(project: &str, peer_user: &str, peer_host: &str) -> Result<()> {
let remote = format!("{}@{}", peer_user, peer_host);
let out = std::process::Command::new("ssh")
.args([&remote, "claw-store", "pull", project])
.output()
.context("ssh to peer")?;
if !out.status.success() {
bail!("peer pull failed: {}", String::from_utf8_lossy(&out.stderr));
}
Ok(())
}
fn project_label(warm_path: &Path) -> String {
let parts: Vec<_> = warm_path.components().rev().take(2).collect();
let parts: Vec<_> = parts.into_iter().rev().collect();
parts.iter()
.map(|c| c.as_os_str().to_string_lossy())
.collect::<Vec<_>>()
.join("/")
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::NamedTempFile;
#[test]
fn test_sync_queue_enqueue_dedup() {
let mut q = SyncQueue::default();
q.enqueue("quantumclaw/quantum-pulse", Path::new("/slab/projects/quantumclaw/quantum-pulse"));
q.enqueue("quantumclaw/quantum-pulse", Path::new("/slab/projects/quantumclaw/quantum-pulse"));
assert_eq!(q.jobs.len(), 1);
}
#[test]
fn test_sync_queue_roundtrip() {
let mut q = SyncQueue::default();
q.enqueue("redclaw/claw-mesh", Path::new("/slab/projects/redclaw/claw-mesh"));
let f = NamedTempFile::new().unwrap();
q.save(f.path()).unwrap();
let loaded = SyncQueue::load(f.path()).unwrap();
assert_eq!(loaded.jobs.len(), 1);
assert_eq!(loaded.jobs[0].project, "redclaw/claw-mesh");
}
#[test]
fn test_sync_queue_drops_after_max_attempts() {
let mut q = SyncQueue::default();
q.enqueue("redclaw/claw-mesh", Path::new("/slab/projects/redclaw/claw-mesh"));
q.jobs[0].attempts = 48;
// Simulate a failed notify by direct manipulation
let mut remaining = Vec::new();
for mut job in q.jobs.drain(..) {
job.attempts += 1;
if job.attempts < 48 {
remaining.push(job);
}
}
q.jobs = remaining;
assert!(q.jobs.is_empty(), "should be dropped after 48 attempts");
}
}