fix library sync backfill O(N^2) hotspots on both sides of the protocol

Addresses #3058 (shared models, content_identity) and #3060 (device-owned
entries). The progressive slowdown on initial pair came from several
independent hotspots that compound:

Sender side
- PeerLog::get_since now pushes LIMIT into SQL; callers fetch limit + 1 to
  derive has_more. Previously every SharedChangeRequest reloaded and
  parsed the entire remaining shared_changes log in memory before
  truncating, making sender work O(N * batches).
- Entry and content_identity query_for_sync batch FK to UUID conversion
  across the whole batch via a new convert_fks_to_uuids_batch helper —
  one DB round trip per FK type instead of per record per FK.

Schema
- New index idx_entries_indexed_at_uuid backing the (indexed_at, uuid)
  cursor in Entry::query_for_sync. Without it every batch request fell
  back to a full-table scan.

Receiver side
- New task-local in_backfill scope wraps the backfill apply phases.
  Entry::apply_state_change uses it to skip per-entry entry_closure
  rebuild; the existing post_backfill_rebuild pass does a single bulk
  rebuild at the end. emit_batch_resource_events short-circuits during
  backfill and the coordinator emits one Event::Refresh after post-
  backfill rebuild so the UI invalidates cached views.
- Entry FK resolution in backfill splits mappings into self-referential
  (parent_id) and the rest. Non-self FKs resolve in one batch across the
  whole batch; only parent_id still runs per-entry so children can see
  just-inserted parents.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jamiepine

core/src/domain/resource_manager.rs

@@ -269,6 +269,19 @@ impl ResourceManager {
 		resource_type: &str,
 		resource_ids: Vec<Uuid>,
 	) -> Result<()> {
+		// During backfill we skip per-record fan-out. For models like
+		// content_identity each UUID triggers 2 DB queries via dependency
+		// routing, which dominates apply time on large libraries. The backfill
+		// coordinator emits a single coarse invalidation after the scope ends.
+		if crate::infra::sync::is_in_backfill() {
+			tracing::trace!(
+				resource_type = %resource_type,
+				count = resource_ids.len(),
+				"Skipping per-record resource event emission during backfill"
+			);
+			return Ok(());
+		}
+
 		// For now, delegate to single-resource handler
 		// In future, could optimize by batching virtual resource construction
 		self.emit_resource_events(resource_type, resource_ids).await

core/src/infra/db/entities/content_identity.rs

@@ -246,36 +246,55 @@ impl Syncable for Model {
 
 		let results = query.all(db).await?;
 
-		let mut sync_results = Vec::new();
+		let mut sync_results: Vec<(Uuid, serde_json::Value, chrono::DateTime<chrono::Utc>)> =
+			Vec::with_capacity(results.len());
 		for content in results {
-			if content.uuid.is_none() {
-				continue;
-			}
+			let uuid = match content.uuid {
+				Some(u) => u,
+				None => continue,
+			};
 
-			let mut json = match content.to_sync_json() {
+			let json = match content.to_sync_json() {
 				Ok(j) => j,
 				Err(e) => {
 					tracing::warn!(error = %e, content_hash = %content.content_hash, "Failed to serialize content_identity for sync");
 					continue;
 				}
 			};
 
-			// Convert FK to UUID for cross-device compatibility
-			for fk in Self::foreign_key_mappings() {
-				if let Err(e) =
-					crate::infra::sync::fk_mapper::convert_fk_to_uuid(&mut json, &fk, db).await
+			sync_results.push((uuid, json, content.last_verified_at));
+		}
+
+		// Batch FK → UUID conversion across the whole batch: one DB round trip
+		// per FK type instead of one per (record × FK).
+		let fk_mappings = Self::foreign_key_mappings();
+		if !fk_mappings.is_empty() && !sync_results.is_empty() {
+			let mut payloads: Vec<serde_json::Value> =
+				sync_results.iter().map(|(_, json, _)| json.clone()).collect();
+
+			for fk in &fk_mappings {
+				if let Err(e) = crate::infra::sync::fk_mapper::convert_fks_to_uuids_batch(
+					&mut payloads,
+					fk,
+					db,
+				)
+				.await
 				{
 					tracing::warn!(
 						error = %e,
-						uuid = %content.uuid.unwrap(),
 						fk_field = fk.local_field,
-						"Failed to convert FK to UUID, skipping content_identity"
+						"Batch FK conversion failed for content_identity"
 					);
-					continue;
+					return Err(sea_orm::DbErr::Custom(format!(
+						"ContentIdentity FK batch conversion failed: {}",
+						e
+					)));
 				}
 			}
 
-			sync_results.push((content.uuid.unwrap(), json, content.last_verified_at));
+			for ((_, json, _), resolved) in sync_results.iter_mut().zip(payloads.into_iter()) {
+				*json = resolved;
+			}
 		}
 
 		Ok(sync_results)

core/src/infra/db/entities/entry.rs

@@ -292,16 +292,16 @@ impl crate::infra::sync::Syncable for Model {
 				std::collections::HashMap::new()
 			};
 
-		// Convert to sync format with FK mapping
-		let mut sync_results = Vec::new();
+		// Serialize each row to JSON with its UUID and timestamp.
+		let mut staged: Vec<(Uuid, serde_json::Value, chrono::DateTime<chrono::Utc>)> =
+			Vec::with_capacity(results.len());
 
 		for entry in results {
 			let uuid = match entry.uuid {
 				Some(u) => u,
-				None => continue, // Skip entries without UUIDs
+				None => continue,
 			};
 
-			// Serialize to JSON
 			let mut json = match entry.to_sync_json() {
 				Ok(j) => j,
 				Err(e) => {
@@ -311,9 +311,8 @@ impl crate::infra::sync::Syncable for Model {
 			};
 
 			// For directories, include the absolute path from directory_paths
-			// This ensures receiving devices get identical paths for universal addressing
+			// so receiving devices get identical paths for universal addressing.
 			if entry.kind == 1 {
-				// Directory
 				if let Some(path) = directory_paths_map.get(&entry.id) {
 					if let Some(obj) = json.as_object_mut() {
 						obj.insert(
@@ -324,27 +323,43 @@ impl crate::infra::sync::Syncable for Model {
 				}
 			}
 
-			// Convert FK integer IDs to UUIDs
-			for fk in <Model as Syncable>::foreign_key_mappings() {
-				if let Err(e) =
-					crate::infra::sync::fk_mapper::convert_fk_to_uuid(&mut json, &fk, db).await
+			let timestamp = entry.indexed_at.unwrap_or(entry.modified_at);
+			staged.push((uuid, json, timestamp));
+		}
+
+		// Batch-convert FK integer IDs to UUIDs one FK type at a time across
+		// the whole batch — single DB round trip per FK, not per record × FK.
+		let fk_mappings = <Model as Syncable>::foreign_key_mappings();
+		if !fk_mappings.is_empty() && !staged.is_empty() {
+			let mut payloads: Vec<serde_json::Value> =
+				staged.iter().map(|(_, json, _)| json.clone()).collect();
+
+			for fk in &fk_mappings {
+				if let Err(e) = crate::infra::sync::fk_mapper::convert_fks_to_uuids_batch(
+					&mut payloads,
+					fk,
+					db,
+				)
+				.await
 				{
 					tracing::warn!(
 						error = %e,
-						uuid = %uuid,
 						fk_field = fk.local_field,
-						"Failed to convert FK to UUID, skipping entry"
+						"Batch FK conversion failed for entries"
 					);
-					continue;
+					return Err(sea_orm::DbErr::Custom(format!(
+						"Entry FK batch conversion failed: {}",
+						e
+					)));
 				}
 			}
 
-			// Use indexed_at for checkpoint/watermark tracking, fallback to modified_at if NULL
-			let timestamp = entry.indexed_at.unwrap_or(entry.modified_at);
-			sync_results.push((uuid, json, timestamp));
+			for ((_, json, _), resolved) in staged.iter_mut().zip(payloads.into_iter()) {
+				*json = resolved;
+			}
 		}
 
-		Ok(sync_results)
+		Ok(staged)
 	}
 
 	/// Apply state change - already implemented in Model impl block below
@@ -564,10 +579,12 @@ impl Model {
 			inserted.id
 		};
 
-		// Rebuild entry_closure for this synced entry
-		// Without this, the entry only has a self-reference and cannot be queried
-		// for descendants, breaking subtree operations, location scoping, etc.
-		Self::rebuild_entry_closure(entry_id, parent_id, db).await?;
+		// Rebuild entry_closure for this synced entry, unless we're inside a
+		// backfill apply loop — in that case the post_backfill_rebuild hook
+		// does a single bulk rebuild at the end, so per-entry work is wasted.
+		if !crate::infra::sync::is_in_backfill() {
+			Self::rebuild_entry_closure(entry_id, parent_id, db).await?;
+		}
 
 		// If this is a directory, create or update its entry in the directory_paths table
 		if EntryKind::from(kind) == EntryKind::Directory {

core/src/infra/db/migration/m20260417_000001_add_entries_sync_cursor_index.rs

@@ -0,0 +1,36 @@
+//! Add composite index on entries(indexed_at, uuid) to back the device-owned
+//! sync cursor.
+//!
+//! `Entry::query_for_sync` paginates by `ORDER BY indexed_at ASC, uuid ASC`
+//! with a tie-breaker filter of the same shape. Without this index, SQLite
+//! does a full table scan per batch request — O(N) per batch, O(N^2) across
+//! an initial backfill of a large library.
+
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+	async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+		manager
+			.get_connection()
+			.execute_unprepared(
+				"CREATE INDEX IF NOT EXISTS idx_entries_indexed_at_uuid \
+				 ON entries(indexed_at, uuid)",
+			)
+			.await?;
+
+		Ok(())
+	}
+
+	async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+		manager
+			.get_connection()
+			.execute_unprepared("DROP INDEX IF EXISTS idx_entries_indexed_at_uuid")
+			.await?;
+
+		Ok(())
+	}
+}

core/src/infra/db/migration/mod.rs

@@ -38,6 +38,7 @@ mod m20260104_000001_replace_device_id_with_volume_id;
 mod m20260105_000001_add_volume_id_to_locations;
 mod m20260114_000001_fix_search_index_include_directories;
 mod m20260123_000001_remove_legacy_sync_columns;
+mod m20260417_000001_add_entries_sync_cursor_index;
 
 pub struct Migrator;
 
@@ -81,6 +82,7 @@ impl MigratorTrait for Migrator {
 			Box::new(m20260105_000001_add_volume_id_to_locations::Migration),
 			Box::new(m20260114_000001_fix_search_index_include_directories::Migration),
 			Box::new(m20260123_000001_remove_legacy_sync_columns::Migration),
+			Box::new(m20260417_000001_add_entries_sync_cursor_index::Migration),
 		]
 	}
 }

core/src/infra/sync/backfill_context.rs

@@ -0,0 +1,26 @@
+//! Task-local flag identifying code running inside a backfill apply loop.
+//!
+//! A few per-record hooks (closure table rebuild, resource event emission)
+//! are redundant during backfill because the coordinator does bulk work at
+//! the end. Models check this flag to skip that per-record work.
+
+tokio::task_local! {
+	static IN_BACKFILL: ();
+}
+
+/// Run `fut` with the in-backfill flag set. Nested scopes are allowed.
+pub async fn in_backfill<F, T>(fut: F) -> T
+where
+	F: std::future::Future<Output = T>,
+{
+	if is_in_backfill() {
+		fut.await
+	} else {
+		IN_BACKFILL.scope((), fut).await
+	}
+}
+
+/// True when the current task is inside an `in_backfill` scope.
+pub fn is_in_backfill() -> bool {
+	IN_BACKFILL.try_with(|_| ()).is_ok()
+}

core/src/infra/sync/fk_mapper.rs

@@ -102,6 +102,85 @@ pub async fn convert_fk_to_uuid(
 	Ok(())
 }
 
+/// Batch convert local integer FKs to UUIDs across multiple records.
+///
+/// Same contract as [`convert_fk_to_uuid`] but one DB round trip per FK type
+/// instead of one per (record × FK). Records that reference a missing target
+/// are left with their original local-id field intact; callers log/skip as
+/// needed. Called from `query_for_sync` implementations on the sync sender.
+pub async fn convert_fks_to_uuids_batch(
+	records: &mut [Value],
+	fk: &FKMapping,
+	db: &DatabaseConnection,
+) -> Result<()> {
+	if records.is_empty() {
+		return Ok(());
+	}
+
+	let uuid_field = fk.uuid_field_name();
+
+	// Collect all local IDs we need to resolve.
+	let mut ids_to_lookup: HashSet<i32> = HashSet::new();
+	for json in records.iter() {
+		match json.get(fk.local_field) {
+			Some(v) if v.is_null() => { /* null FK, handled below */ }
+			Some(v) => {
+				if let Some(id) = v.as_i64() {
+					ids_to_lookup.insert(id as i32);
+				}
+			}
+			None => { /* field absent, handled below */ }
+		}
+	}
+
+	let id_to_uuid = if ids_to_lookup.is_empty() {
+		HashMap::new()
+	} else {
+		batch_lookup_uuids_for_local_ids(fk.target_table, ids_to_lookup, db).await?
+	};
+
+	for json in records.iter_mut() {
+		let local_field_value = json.get(fk.local_field).cloned();
+
+		match local_field_value {
+			Some(v) if v.is_null() => {
+				json[&uuid_field] = Value::Null;
+				if let Some(obj) = json.as_object_mut() {
+					obj.remove(fk.local_field);
+				}
+			}
+			Some(v) => {
+				if let Some(id) = v.as_i64() {
+					match id_to_uuid.get(&(id as i32)) {
+						Some(uuid) => {
+							json[&uuid_field] = json!(uuid.to_string());
+							if let Some(obj) = json.as_object_mut() {
+								obj.remove(fk.local_field);
+							}
+						}
+						None => {
+							// Target record missing locally — leave FK field intact
+							// so the caller can decide whether to skip or warn.
+							tracing::warn!(
+								fk_field = fk.local_field,
+								target_table = fk.target_table,
+								id = id,
+								"FK target not found during batch conversion; skipping record"
+							);
+						}
+					}
+				}
+			}
+			None => {
+				// Field absent — treat as null for sync payloads.
+				json[&uuid_field] = Value::Null;
+			}
+		}
+	}
+
+	Ok(())
+}
+
 /// Look up UUID for a local integer ID via the registry
 async fn lookup_uuid_for_local_id(
 	table: &str,

core/src/infra/sync/mod.rs

@@ -10,6 +10,7 @@
 //! - Checkpoint persistence for resumable backfill
 //!
 
+pub mod backfill_context;
 pub mod checkpoints;
 pub mod config;
 pub mod dependency_graph;
@@ -27,6 +28,7 @@ pub mod transaction;
 pub mod transport;
 pub mod watermarks;
 
+pub use backfill_context::{in_backfill, is_in_backfill};
 pub use checkpoints::{BackfillCheckpoint, BackfillCheckpointStore, CheckpointError};
 pub use config::{
 	BatchingConfig, MonitoringConfig, NetworkConfig, PruningStrategy, RetentionConfig, SyncConfig,
@@ -42,8 +44,8 @@ pub use event_log::{
 	SyncEventLogger, SyncEventQuery, SyncEventType,
 };
 pub use fk_mapper::{
-	batch_map_sync_json_to_local, convert_fk_to_uuid, map_sync_json_to_local, BatchFkMapResult,
-	FKMapping,
+	batch_map_sync_json_to_local, convert_fk_to_uuid, convert_fks_to_uuids_batch,
+	map_sync_json_to_local, BatchFkMapResult, FKMapping,
 };
 pub use hlc::{HLCGenerator, HLC};
 pub use peer_log::{ChangeType, PeerLog, PeerLogError, SharedChangeEntry};