sync: fetch content, not just metadata, from Ensemble screen's per-peer Sync

The Ensemble/EnsembleScreen "Sync" button only ever ran the catalogue pass
(sync_from_peer), which exchanges document metadata but never blob bytes.
Content only moved via the separate scheduler's content pass (periodic
timer or the Settings "Run now" button), so setting a node's SyncPolicy to
Mirror and hitting the peer-row Sync button looked like it did nothing for
content (e.g. comic page images never showing up on a newly-mirrored node).

Add scheduler::run_content_passes_for_peer, scoping the existing
catalogue-fetch and content passes to a single peer, and call it from both
the button's sync() handler (gui-app/src/ensemble.rs) and the Android FFI's
SyncNode::sync_documents after the catalogue sync completes. sync_documents
now returns a SyncDocumentsResult record (documents_synced + blobs_fetched)
instead of a bare doc count so both platforms can report what actually
moved.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0134Qxsmkvtr8hSGkhbT1nTg

android/app/src/main/java/com/gregshuflin/synchronicity/NativeLib.kt

@@ -39,6 +39,7 @@ import uniffi.synchronicity.ReclaimableBytesItem
 import uniffi.synchronicity.RemoteDocumentSyncPolicy
 import uniffi.synchronicity.SchedulerRunSummary
 import uniffi.synchronicity.SchedulerSettings
+import uniffi.synchronicity.SyncDocumentsResult
 import uniffi.synchronicity.SyncException
 import uniffi.synchronicity.SyncNode
 import uniffi.synchronicity.SyncPolicy
@@ -347,7 +348,9 @@ class NativeLib {
         node().pingNode(nodeIdHex, timeoutSecs.toUInt())
     }
 
-    fun syncDocuments(nodeIdHex: String): Result<UInt> = ffi { node().syncDocuments(nodeIdHex) }
+    fun syncDocuments(nodeIdHex: String): Result<SyncDocumentsResult> = ffi {
+        node().syncDocuments(nodeIdHex)
+    }
 
     // Pull this document's blobs + artifact values, and those of every descendant
     // document, from ensemble peers into the local node, persisting everything

android/app/src/main/java/com/gregshuflin/synchronicity/ensemble/NodeControls.kt

@@ -137,12 +137,14 @@ internal fun launchSync(
     scope.launch {
         syncStates[nodeIdHex] = SyncState(pending = true)
         withContext(Dispatchers.IO) { nativeLib.syncDocuments(nodeIdHex) }
-            .onSuccess { count ->
+            .onSuccess { result ->
                 // Format via shared Rust so the live value matches the persisted
                 // one shown after a reload.
                 val nowSecs = System.currentTimeMillis() / MILLIS_PER_SECOND
                 val ts = nativeLib.formatSeenTimestamp(nowSecs, zone.id)
-                syncStates[nodeIdHex] = SyncState(ok = "synced $count doc(s)", lastSynced = ts)
+                val message =
+                    "synced ${result.documentsSynced} doc(s), ${result.blobsFetched} blob(s) fetched"
+                syncStates[nodeIdHex] = SyncState(ok = message, lastSynced = ts)
             }
             .onFailure { syncStates[nodeIdHex] = SyncState(error = it.message ?: "sync failed") }
     }

rust/gui-app/src/ensemble.rs

@@ -20,6 +20,7 @@ use synchronicity::jobs::{JobKind, JobOrigin, JobProgress, JobSubject};
 use synchronicity::node::{NodePubkey, NodeRuntime, do_bootstrap_retry};
 use synchronicity::p2p::P2PNode;
 use synchronicity::p2p::doc_sync::{SyncProgress, SyncRequest, sync_from_peer};
+use synchronicity::scheduler;
 use synchronicity::stores::Store;
 use tokio::runtime::Handle;
 use tokio_stream::StreamExt;
@@ -231,6 +232,7 @@ impl EnsembleView {
         let endpoint = self.node_runtime.lock().unwrap().as_ref().map(|rt| rt.p2p.endpoint());
         let rt = self.rt.clone();
         let store = Arc::clone(&self.store);
+        let job_runner = Arc::clone(&self.job_runner);
         let probe = target.clone();
         cx.spawn(async move |this, cx| {
             let result = cx
@@ -332,12 +334,36 @@ impl EnsembleView {
                                 store.peer_node_store.record_sync(tx, &bytes, Timestamp(now))
                             });
                         }
-                        let message = if skipped_total > 0 {
-                            format!(
-                                "synced {total} doc(s), {skipped_total} skipped (peer newer — upgrade to get them)"
-                            )
-                        } else {
-                            format!("synced {total} doc(s)")
+
+                        // Catalogue metadata is now up to date with this peer; also pull
+                        // whatever content (blobs) this node's own `SyncPolicy` wants
+                        // and this peer holds — otherwise "Sync" would only ever move
+                        // metadata, never the `Mirror`-policy bytes the button implies.
+                        // Best-effort: a fetch failure here shouldn't turn an otherwise-
+                        // successful catalogue sync into an error, so it's swallowed and
+                        // just left out of the summary rather than propagated with `?`.
+                        let content_summary = rt
+                            .block_on(scheduler::run_content_passes_for_peer(
+                                &ep,
+                                Arc::clone(&store),
+                                &job_runner,
+                                peer,
+                            ))
+                            .map(|(_catalogue_fetch, content)| content);
+
+                        let message = match (skipped_total, content_summary) {
+                            (0, None) => format!("synced {total} doc(s)"),
+                            (0, Some(content)) => format!(
+                                "synced {total} doc(s), {} blob(s) fetched",
+                                content.blobs_fetched
+                            ),
+                            (skipped, None) => format!(
+                                "synced {total} doc(s), {skipped} skipped (peer newer — upgrade to get them)"
+                            ),
+                            (skipped, Some(content)) => format!(
+                                "synced {total} doc(s), {skipped} skipped (peer newer — upgrade to get them), {} blob(s) fetched",
+                                content.blobs_fetched
+                            ),
                         };
                         Ok((message, last_synced))
                     }

rust/lib/src/ffi.rs

@@ -1114,6 +1114,16 @@ pub struct ReclaimableBytesItem {
     pub bytes: u64,
 }
 
+/// Result of [`SyncNode::sync_documents`]: the catalogue pass's document
+/// count plus whatever the follow-on content pass fetched from the same
+/// peer, so the Ensemble screen's "Sync" button can report both instead of
+/// implying (as a bare document count did) that content moved too.
+#[derive(uniffi::Record)]
+pub struct SyncDocumentsResult {
+    pub documents_synced: u32,
+    pub blobs_fetched: u32,
+}
+
 /// Summary of one [`SyncNode::run_scheduler_passes`] tick, for display/logging
 /// (the Android counterpart of `scheduler_driver.rs`'s log line).
 #[derive(uniffi::Record)]
@@ -1892,7 +1902,7 @@ impl SyncNode {
         Ok(rtt.as_secs_f64() * 1000.0)
     }
 
-    pub fn sync_documents(&self, node_id_hex: String) -> Result<u32, SyncError> {
+    pub fn sync_documents(&self, node_id_hex: String) -> Result<SyncDocumentsResult, SyncError> {
         use tokio_stream::StreamExt;
 
         let (endpoint, rt_handle, store) = {
@@ -1997,7 +2007,24 @@ impl SyncNode {
             });
         }
 
-        Ok(total)
+        // Catalogue metadata is now current with this peer; also pull
+        // whatever content this node's own `SyncPolicy` wants (e.g. `Mirror`)
+        // and this peer holds — otherwise "Sync" would only ever move
+        // metadata, never blob bytes, no matter the policy (see
+        // `scheduler::run_content_passes_for_peer`'s doc comment). Best-effort:
+        // failing to fetch content shouldn't turn an otherwise-successful
+        // catalogue sync into an error.
+        let blobs_fetched = rt_handle
+            .block_on(scheduler::run_content_passes_for_peer(
+                &endpoint,
+                Arc::clone(&store),
+                &self.job_runner,
+                peer,
+            ))
+            .map(|(_catalogue_fetch, content)| content.blobs_fetched)
+            .unwrap_or(0);
+
+        Ok(SyncDocumentsResult { documents_synced: total, blobs_fetched })
     }
 
     pub fn stop_p2p_node(&self) -> Result<(), SyncError> {

rust/lib/src/scheduler/mod.rs

@@ -89,11 +89,38 @@ pub async fn run_all_passes(
     Some(SchedulerPassResult { catalogue, catalogue_fetch, content })
 }
 
+/// Run all three scheduler passes against a single peer, rather than every
+/// reachable ensemble peer — the Ensemble screen's per-peer "Sync" button
+/// (and its Android equivalent) wants "sync everything with *this* node",
+/// not a full-ensemble [`run_all_passes`] run. Unlike that function, this
+/// isn't wrapped in a `BackgroundSync` job (the caller already has its own
+/// `CatalogueSync` job for UI progress on pass 1); passes 2 and 3 create
+/// their own jobs as they go (`run_content_pass` starts one `ContentFetch`
+/// job per document), so progress for those is visible via the Jobs screen.
+///
+/// Returns `None` if `peer` isn't a member of this node's own ensemble (the
+/// screen listing it implies it should be, but a caller shouldn't panic if
+/// membership changed underneath it) — in which case the caller should fall
+/// back to reporting only the catalogue pass it already ran.
+pub async fn run_content_passes_for_peer(
+    endpoint: &Endpoint,
+    store: Arc<Store>,
+    job_runner: &JobRunner,
+    peer: NodePubkey,
+) -> Option<(CatalogueFetchPassResult, ContentPassResult)> {
+    let peer_node =
+        reachable_ensemble_peers(&store).into_iter().find(|p| p.node_pubkey == peer.0)?;
+    let peers = [peer_node];
+    let catalogue_fetch = run_catalogue_fetch_pass(endpoint, &store, &peers).await;
+    let content = run_content_pass(endpoint, Arc::clone(&store), job_runner, &peers).await;
+    Some((catalogue_fetch, content))
+}
+
 /// Every peer in this node's ensemble known via gossip, per
 /// [`PeerNodeStore::get_peer_nodes_for_ensemble`](crate::stores::peer_nodes::PeerNodeStore::get_peer_nodes_for_ensemble).
 /// Empty (rather than an error) if this node has no ensemble membership yet
 /// or the membership can't be loaded — nothing to sync against either way.
-pub(crate) fn reachable_ensemble_peers(store: &Store) -> Vec<PeerNode> {
+pub fn reachable_ensemble_peers(store: &Store) -> Vec<PeerNode> {
     let Ok(Some(membership)) =
         store.settings_store.batch(|tx| store.settings_store.load_ensemble_membership(tx))
     else {