sync: never silently drop; branch-aware pull; expose queue depth in /api/status

Three fixes bundled — all defensive around the architect↔tank dev flow:

1. drain_sync_queue no longer drops jobs after 48 attempts. The
   previous cutoff (~4h at 5-min drain intervals) silently lost every
   pending sync whenever the peer was down longer than that — the
   exact scenario the user hit last week when architect was offline.
   Now jobs are retried indefinitely; past SYNC_STUCK_ATTEMPTS (12,
   ~1h) they escalate from WARN to ERROR and appear in /api/status
   under sync_queue_stuck so the operator sees them.

2. pull_project is now branch-aware. Old behavior was 'git pull
   --ff-only' on whatever branch happened to be checked out. If the
   peer was on main but the sender committed on develop, the peer
   never picked up the new branch's ref. New behavior:
     - git fetch --all --prune --tags first (gets every branch)
     - fast-forward the checked-out branch if it has an upstream
     - fast-forward every OTHER branch that has an upstream via
       update-ref, without switching HEAD — so main can advance while
       the operator is off hacking on feature/x.
   Diverged branches are left alone (loud error, not silent merge).
   Detached HEAD is skipped after fetch.

3. NodeStatus gains sync_queue_depth + sync_queue_stuck fields, wired
   into both handle_status (GET /api/status) and the SSE stream. The
   dashboard now has ground truth for 'is anything backing up?'

Tests updated: the old test_sync_queue_drops_after_max_attempts is
replaced by test_sync_queue_never_drops_stuck_jobs (invariant: never
drops) and test_sync_queue_stuck_count_threshold (invariant: counts
jobs past the escalation threshold).

claw-store/src/serve.rs

@@ -46,6 +46,12 @@ pub struct NodeStatus {
     pub zfs_pool_state: String,
     pub active_project_count: usize,
     pub daemon_uptime_secs: u64,
+    /// Number of git-sync notifications waiting for the peer to accept.
+    pub sync_queue_depth: usize,
+    /// Number of queued jobs that have exceeded the escalation threshold
+    /// (~1h of retries). Non-zero means a peer has been unreachable long
+    /// enough that we've stopped hiding it in WARN-level logs.
+    pub sync_queue_stuck: usize,
 }
 
 #[derive(Serialize)]
@@ -351,6 +357,10 @@ async fn handle_status(State(state): State<Arc<AppState>>) -> Json<NodeStatus> {
         crate::config::NodeRole::Secondary => "secondary",
     };
 
+    // Cheap file read; the queue is bounded by pending projects, not job
+    // history, so the file stays small in practice.
+    let queue = SyncQueue::load(&SyncQueue::default_path()).unwrap_or_default();
+
     Json(NodeStatus {
         node_name: cfg.node.name.clone(),
         role: role.to_string(),
@@ -367,6 +377,8 @@ async fn handle_status(State(state): State<Arc<AppState>>) -> Json<NodeStatus> {
         zfs_pool_state,
         active_project_count: manifest.projects.len(),
         daemon_uptime_secs: daemon_uptime_secs(),
+        sync_queue_depth: queue.depth(),
+        sync_queue_stuck: queue.stuck_count(crate::sync::SYNC_STUCK_ATTEMPTS),
     })
 }
 
@@ -580,6 +592,7 @@ async fn handle_events(
             crate::config::NodeRole::Secondary => "secondary",
         };
 
+        let queue = SyncQueue::load(&SyncQueue::default_path()).unwrap_or_default();
         let status = NodeStatus {
             node_name: cfg.node.name.clone(),
             role: role.to_string(),
@@ -596,6 +609,8 @@ async fn handle_events(
             zfs_pool_state,
             active_project_count: manifest.projects.len(),
             daemon_uptime_secs: daemon_uptime_secs(),
+            sync_queue_depth: queue.depth(),
+            sync_queue_stuck: queue.stuck_count(crate::sync::SYNC_STUCK_ATTEMPTS),
         };
 
         // v0.2.1 — single-source the project list builder. The previous

claw-store/src/sync.rs

@@ -46,8 +46,25 @@ impl SyncQueue {
     pub fn default_path() -> PathBuf {
         PathBuf::from("/var/lib/claw-store/sync-queue.toml")
     }
+
+    /// Total number of pending jobs.
+    pub fn depth(&self) -> usize {
+        self.jobs.len()
+    }
+
+    /// Jobs that have exceeded the escalation threshold — surfaced in
+    /// `/api/status` so a peer that's been unreachable for hours becomes
+    /// visible in the dashboard rather than silently accumulating retries.
+    pub fn stuck_count(&self, escalate_after: u32) -> usize {
+        self.jobs.iter().filter(|j| j.attempts >= escalate_after).count()
+    }
 }
 
+/// A job that hits this many attempts is escalated to ERROR-level logging
+/// and reported in `/api/status` as "stuck". Retries continue indefinitely
+/// — the tool never silently drops a sync — but the operator now sees it.
+pub const SYNC_STUCK_ATTEMPTS: u32 = 12; // ~1h at 5-min drain intervals
+
 /// Push local commits to origin then notify peer to pull.
 pub fn sync_project(
     warm_path: &Path,
@@ -60,25 +77,141 @@ pub fn sync_project(
 }
 
 /// Pull latest from origin into warm_path.
+///
+/// This is branch-aware, not just "pull the current branch":
+///   1. `git fetch --all --prune --tags` picks up every remote branch and
+///      drops refs that are gone upstream. Without this, if the peer is
+///      checked out on `main` but the sender committed on `develop`, the
+///      peer never sees the new branch's ref.
+///   2. If HEAD is a real branch (not detached), we try a fast-forward
+///      of the checked-out branch against its upstream. Non-ff diffs bail
+///      loud instead of silently merging.
+///   3. Every other local branch that has an upstream and can fast-forward
+///      is advanced *without switching branches* via `update-ref` — so a
+///      developer working on `feature/x` on the peer still gets `main`
+///      pulled up to date behind their back.
 pub fn pull_project(warm_path: &Path) -> Result<()> {
     if !warm_path.exists() {
         bail!("warm path does not exist: {}", warm_path.display());
     }
     let path_str = warm_path.to_str()
         .with_context(|| format!("non-UTF-8 path: {}", warm_path.display()))?;
-    let out = std::process::Command::new("git")
+
-        .args(["-C", path_str, "pull", "--ff-only"])
+    // Step 1: fetch every ref
+    let fetch = std::process::Command::new("git")
+        .args(["-C", path_str, "fetch", "--all", "--prune", "--tags"])
         .output()
-        .context("running git pull")?;
+        .context("running git fetch")?;
-    if !out.status.success() {
+    if !fetch.status.success() {
-        bail!("git pull failed: {}", String::from_utf8_lossy(&out.stderr));
+        bail!("git fetch failed: {}", String::from_utf8_lossy(&fetch.stderr));
     }
-    tracing::info!("pulled {}: {}", project_label(warm_path),
+
-        String::from_utf8_lossy(&out.stdout).trim());
+    // Step 2: detect current branch (empty output ⇒ detached HEAD)
+    let head = std::process::Command::new("git")
+        .args(["-C", path_str, "symbolic-ref", "-q", "--short", "HEAD"])
+        .output()
+        .context("running git symbolic-ref")?;
+    let current_branch = if head.status.success() {
+        String::from_utf8_lossy(&head.stdout).trim().to_string()
+    } else {
+        // Detached HEAD (rebase, bisect, mid-checkout of a tag). Leave the
+        // working tree alone — the fetch already gave us the refs; the
+        // developer picks their next move manually.
+        tracing::info!("pull {}: detached HEAD, skipping ff merge (fetch only)",
+            project_label(warm_path));
+        return Ok(());
+    };
+
+    // Step 3: fast-forward the checked-out branch against its upstream. This
+    // uses `merge --ff-only` so a genuine divergence errors loudly instead
+    // of quietly interleaving commits.
+    let merge = std::process::Command::new("git")
+        .args(["-C", path_str, "merge", "--ff-only", "@{u}"])
+        .output()
+        .context("running git merge --ff-only")?;
+    if !merge.status.success() {
+        let err = String::from_utf8_lossy(&merge.stderr);
+        // "no upstream configured" is not fatal — the branch just doesn't
+        // track anything remote (feature branch created locally, etc.).
+        if err.contains("no upstream") || err.contains("does not have") {
+            tracing::info!("pull {}: branch '{}' has no upstream, fetch only",
+                project_label(warm_path), current_branch);
+        } else {
+            bail!("git merge --ff-only failed on branch '{}': {}", current_branch, err);
+        }
+    } else {
+        let out = String::from_utf8_lossy(&merge.stdout);
+        if out.trim() != "Already up to date." {
+            tracing::info!("pull {} ({}): {}",
+                project_label(warm_path), current_branch, out.trim());
+        }
+    }
+
+    // Step 4: for every OTHER local branch that has an upstream, try to
+    // fast-forward its ref in place (no checkout switch). This is what
+    // lets `main` on the peer stay current while the operator is off
+    // hacking on `feature/x`. Failures here are best-effort — a branch
+    // that's diverged just stays where it is.
+    if let Ok(branches) = list_local_branches(path_str) {
+        for branch in branches {
+            if branch == current_branch { continue; }
+            let _ = fast_forward_branch(path_str, &branch);
+        }
+    }
+
     Ok(())
 }
 
-/// Process pending sync queue — retry each job, drop successes, bump attempt counter.
+fn list_local_branches(path_str: &str) -> Result<Vec<String>> {
+    let out = std::process::Command::new("git")
+        .args(["-C", path_str, "for-each-ref", "--format=%(refname:short)", "refs/heads/"])
+        .output()
+        .context("git for-each-ref")?;
+    if !out.status.success() {
+        bail!("for-each-ref failed: {}", String::from_utf8_lossy(&out.stderr));
+    }
+    Ok(String::from_utf8_lossy(&out.stdout)
+        .lines()
+        .map(str::to_string)
+        .filter(|s| !s.is_empty())
+        .collect())
+}
+
+fn fast_forward_branch(path_str: &str, branch: &str) -> Result<()> {
+    // Resolve upstream ref for this branch, if any.
+    let up = std::process::Command::new("git")
+        .args(["-C", path_str, "rev-parse", "--abbrev-ref", &format!("{branch}@{{u}}")])
+        .output()?;
+    if !up.status.success() { return Ok(()); } // no upstream: skip
+    let upstream = String::from_utf8_lossy(&up.stdout).trim().to_string();
+    if upstream.is_empty() { return Ok(()); }
+
+    // Only advance if local is a strict ancestor of the upstream.
+    let anc = std::process::Command::new("git")
+        .args(["-C", path_str, "merge-base", "--is-ancestor", branch, &upstream])
+        .status()?;
+    if !anc.success() { return Ok(()); } // diverged, leave alone
+
+    // Move the branch ref forward. Won't touch the working tree since we're
+    // not on this branch.
+    let target = std::process::Command::new("git")
+        .args(["-C", path_str, "rev-parse", &upstream])
+        .output()?;
+    let sha = String::from_utf8_lossy(&target.stdout).trim().to_string();
+    if sha.is_empty() { return Ok(()); }
+
+    let _ = std::process::Command::new("git")
+        .args(["-C", path_str, "update-ref", &format!("refs/heads/{branch}"), &sha])
+        .status()?;
+    Ok(())
+}
+
+/// Process pending sync queue — retry each job, drop successes, bump the
+/// attempt counter. Failures are NEVER dropped: the previous behavior was
+/// to give up after 48 attempts (~4h), which silently lost work whenever
+/// the peer was unreachable for longer than that. Instead, jobs that hit
+/// `SYNC_STUCK_ATTEMPTS` escalate from WARN to ERROR log level and appear
+/// in `/api/status` under `sync_queue_stuck` so the operator sees them.
 pub fn drain_sync_queue(
     queue: &mut SyncQueue,
     peer_user: &str,
@@ -91,11 +224,18 @@ pub fn drain_sync_queue(
         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 {}): {:#}",
+                if job.attempts >= SYNC_STUCK_ATTEMPTS {
-                    job.project, job.attempts, e);
+                    tracing::error!(
-                if job.attempts < 48 { // drop after 48 attempts (~4h at 5min intervals)
+                        "sync queue: peer notify STUCK for {} (attempt {}, queued {}): {:#}",
-                    remaining.push(job);
+                        job.project, job.attempts, job.queued_at, e
+                    );
+                } else {
+                    tracing::warn!(
+                        "sync queue: peer notify failed for {} (attempt {}): {:#}",
+                        job.project, job.attempts, e
+                    );
                 }
+                remaining.push(job); // keep retrying — never silently drop
             }
         }
     }
@@ -168,19 +308,27 @@ mod tests {
     }
 
     #[test]
-    fn test_sync_queue_drops_after_max_attempts() {
+    fn test_sync_queue_never_drops_stuck_jobs() {
+        // Previously we dropped jobs after 48 attempts. That silently lost
+        // sync work whenever the peer was down for >4h. The new contract
+        // is: never drop, just escalate. This asserts the invariant.
         let mut q = SyncQueue::default();
         q.enqueue("redclaw/claw-mesh", Path::new("/slab/projects/redclaw/claw-mesh"));
-        q.jobs[0].attempts = 48;
+        q.jobs[0].attempts = 500;
-        // Simulate a failed notify by direct manipulation
+        assert_eq!(q.depth(), 1);
-        let mut remaining = Vec::new();
+        assert_eq!(q.stuck_count(SYNC_STUCK_ATTEMPTS), 1);
-        for mut job in q.jobs.drain(..) {
-            job.attempts += 1;
-            if job.attempts < 48 {
-                remaining.push(job);
     }
-        }
+
-        q.jobs = remaining;
+    #[test]
-        assert!(q.jobs.is_empty(), "should be dropped after 48 attempts");
+    fn test_sync_queue_stuck_count_threshold() {
+        let mut q = SyncQueue::default();
+        q.enqueue("a/one", Path::new("/x/one"));
+        q.enqueue("b/two", Path::new("/x/two"));
+        q.enqueue("c/three", Path::new("/x/three"));
+        q.jobs[0].attempts = 1;
+        q.jobs[1].attempts = SYNC_STUCK_ATTEMPTS;
+        q.jobs[2].attempts = SYNC_STUCK_ATTEMPTS + 30;
+        assert_eq!(q.depth(), 3);
+        assert_eq!(q.stuck_count(SYNC_STUCK_ATTEMPTS), 2);
     }
 }