address PR review feedback: drop unresolved FKs, guard limits

convert_fks_to_uuids_batch now returns the set of record indices whose FK
could not be resolved (missing target row or non-integer value). Callers
in entry::query_for_sync and content_identity::query_for_sync drop those
records from the outgoing sync batch instead of zipping the partially
converted payload back. Previously the sender would ship a record with
its local int field intact and no *_uuid field; the receiver's
map_sync_json_to_local interpreted that as already-resolved and wrote
the sender-local integer directly to the local DB, corrupting the FK.

Also: reject limit == 0 on SharedChangeRequest / get_shared_changes up
front (would have returned 0 rows with has_more = true and spun), and
clamp the SQL LIMIT bind with i64::try_from(..).unwrap_or(i64::MAX)
instead of a wrapping `as i64`.

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

Author: jamiepine

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

@@ -266,35 +266,55 @@ impl Syncable for Model {
 		}
 
 		// Batch FK → UUID conversion across the whole batch: one DB round trip
-		// per FK type instead of one per (record × FK).
+		// per FK type instead of one per (record × FK). Records that fail
+		// resolution are dropped so peers never see a sender-local int in the
+		// payload.
 		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();
+			let mut failed_indices: std::collections::HashSet<usize> =
+				std::collections::HashSet::new();
 
 			for fk in &fk_mappings {
-				if let Err(e) = crate::infra::sync::fk_mapper::convert_fks_to_uuids_batch(
+				match crate::infra::sync::fk_mapper::convert_fks_to_uuids_batch(
 					&mut payloads,
 					fk,
 					db,
 				)
 				.await
 				{
-					tracing::warn!(
-						error = %e,
-						fk_field = fk.local_field,
-						"Batch FK conversion failed for content_identity"
-					);
-					return Err(sea_orm::DbErr::Custom(format!(
-						"ContentIdentity FK batch conversion failed: {}",
-						e
-					)));
+					Ok(failed) => failed_indices.extend(failed),
+					Err(e) => {
+						tracing::warn!(
+							error = %e,
+							fk_field = fk.local_field,
+							"Batch FK conversion failed for content_identity"
+						);
+						return Err(sea_orm::DbErr::Custom(format!(
+							"ContentIdentity FK batch conversion failed: {}",
+							e
+						)));
+					}
 				}
 			}
 
-			for ((_, json, _), resolved) in sync_results.iter_mut().zip(payloads.into_iter()) {
-				*json = resolved;
-			}
+			sync_results = sync_results
+				.into_iter()
+				.zip(payloads.into_iter())
+				.enumerate()
+				.filter_map(|(idx, ((uuid, _, ts), resolved))| {
+					if failed_indices.contains(&idx) {
+						tracing::warn!(
+							uuid = %uuid,
+							"Dropping content_identity with unresolved FK from sync batch"
+						);
+						None
+					} else {
+						Some((uuid, resolved, ts))
+					}
+				})
+				.collect();
 		}
 
 		Ok(sync_results)

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

@@ -329,34 +329,54 @@ impl crate::infra::sync::Syncable for Model {
 
 		// 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.
+		// Any record that fails resolution (missing target, bad value) is
+		// dropped before we return so peers never see a sender-local int.
 		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();
+			let mut failed_indices: std::collections::HashSet<usize> =
+				std::collections::HashSet::new();
 
 			for fk in &fk_mappings {
-				if let Err(e) = crate::infra::sync::fk_mapper::convert_fks_to_uuids_batch(
+				match crate::infra::sync::fk_mapper::convert_fks_to_uuids_batch(
 					&mut payloads,
 					fk,
 					db,
 				)
 				.await
 				{
-					tracing::warn!(
-						error = %e,
-						fk_field = fk.local_field,
-						"Batch FK conversion failed for entries"
-					);
-					return Err(sea_orm::DbErr::Custom(format!(
-						"Entry FK batch conversion failed: {}",
-						e
-					)));
+					Ok(failed) => failed_indices.extend(failed),
+					Err(e) => {
+						tracing::warn!(
+							error = %e,
+							fk_field = fk.local_field,
+							"Batch FK conversion failed for entries"
+						);
+						return Err(sea_orm::DbErr::Custom(format!(
+							"Entry FK batch conversion failed: {}",
+							e
+						)));
+					}
 				}
 			}
 
-			for ((_, json, _), resolved) in staged.iter_mut().zip(payloads.into_iter()) {
-				*json = resolved;
-			}
+			staged = staged
+				.into_iter()
+				.zip(payloads.into_iter())
+				.enumerate()
+				.filter_map(|(idx, ((uuid, _, ts), resolved))| {
+					if failed_indices.contains(&idx) {
+						tracing::warn!(
+							uuid = %uuid,
+							"Dropping entry with unresolved FK from sync batch"
+						);
+						None
+					} else {
+						Some((uuid, resolved, ts))
+					}
+				})
+				.collect();
 		}
 
 		Ok(staged)

core/src/infra/sync/fk_mapper.rs

@@ -104,32 +104,49 @@ pub async fn convert_fk_to_uuid(
 
 /// 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.
+/// Same per-record semantics as [`convert_fk_to_uuid`] but one DB round trip
+/// per FK type instead of one per (record × FK). Records whose target could
+/// not be resolved (missing target row, non-integer value) are reported by
+/// index — callers MUST drop those records rather than ship them. Silently
+/// leaving `local_field` in place would let the receiver's
+/// `map_sync_json_to_local` interpret the sender-local integer as
+/// already-resolved and write it straight into the receiver's DB, corrupting
+/// FKs across devices.
+///
+/// Successful records are mutated in place (`local_field` removed,
+/// `uuid_field` added). The indices of failed records are returned.
 pub async fn convert_fks_to_uuids_batch(
 	records: &mut [Value],
 	fk: &FKMapping,
 	db: &DatabaseConnection,
-) -> Result<()> {
+) -> Result<HashSet<usize>> {
+	let mut failed: HashSet<usize> = HashSet::new();
 	if records.is_empty() {
-		return Ok(());
+		return Ok(failed);
 	}
 
 	let uuid_field = fk.uuid_field_name();
 
-	// Collect all local IDs we need to resolve.
+	// First pass: collect IDs and flag records whose local_field isn't a valid
+	// integer so they can be dropped by the caller.
 	let mut ids_to_lookup: HashSet<i32> = HashSet::new();
-	for json in records.iter() {
+	for (idx, json) in records.iter().enumerate() {
 		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() {
+			None => { /* absent — treated as null below */ }
+			Some(v) if v.is_null() => { /* null — treated as null below */ }
+			Some(v) => match v.as_i64() {
+				Some(id) => {
 					ids_to_lookup.insert(id as i32);
 				}
-			}
-			None => { /* field absent, handled below */ }
+				None => {
+					tracing::warn!(
+						fk_field = fk.local_field,
+						value = %v,
+						"FK field has non-integer value in sync payload; dropping record"
+					);
+					failed.insert(idx);
+				}
+			},
 		}
 	}
 
@@ -139,46 +156,48 @@ pub async fn convert_fks_to_uuids_batch(
 		batch_lookup_uuids_for_local_ids(fk.target_table, ids_to_lookup, db).await?
 	};
 
-	for json in records.iter_mut() {
+	for (idx, json) in records.iter_mut().enumerate() {
+		if failed.contains(&idx) {
+			continue;
+		}
+
 		let local_field_value = json.get(fk.local_field).cloned();
 
 		match local_field_value {
+			None => {
+				json[&uuid_field] = Value::Null;
+			}
 			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"
-							);
+				// Unwrap is safe: first pass validated i64 or flagged as failed.
+				let id = v.as_i64().expect("validated in first pass") as i32;
+				match id_to_uuid.get(&id) {
+					Some(uuid) => {
+						json[&uuid_field] = json!(uuid.to_string());
+						if let Some(obj) = json.as_object_mut() {
+							obj.remove(fk.local_field);
 						}
 					}
+					None => {
+						tracing::warn!(
+							fk_field = fk.local_field,
+							target_table = fk.target_table,
+							id = id,
+							"FK target not found during batch conversion; dropping record"
+						);
+						failed.insert(idx);
+					}
 				}
 			}
-			None => {
-				// Field absent — treat as null for sync payloads.
-				json[&uuid_field] = Value::Null;
-			}
 		}
 	}
 
-	Ok(())
+	Ok(failed)
 }
 
 /// Look up UUID for a local integer ID via the registry

core/src/infra/sync/peer_log.rs

@@ -221,11 +221,14 @@ impl PeerLog {
 		since: Option<HLC>,
 		limit: Option<usize>,
 	) -> Result<Vec<SharedChangeEntry>, PeerLogError> {
-		let query = match (since, limit) {
+		// Clamp to i64::MAX so a ludicrously large usize never wraps into a
+		// negative value when SQLite binds the parameter.
+		let sql_limit = limit.map(|lim| i64::try_from(lim).unwrap_or(i64::MAX));
+		let query = match (since, sql_limit) {
 			(Some(hlc), Some(lim)) => Statement::from_sql_and_values(
 				DbBackend::Sqlite,
 				"SELECT hlc, model_type, record_uuid, change_type, data FROM shared_changes WHERE hlc > ? ORDER BY hlc ASC LIMIT ?",
-				vec![hlc.to_string().into(), (lim as i64).into()],
+				vec![hlc.to_string().into(), lim.into()],
 			),
 			(Some(hlc), None) => Statement::from_sql_and_values(
 				DbBackend::Sqlite,
@@ -235,7 +238,7 @@ impl PeerLog {
 			(None, Some(lim)) => Statement::from_sql_and_values(
 				DbBackend::Sqlite,
 				"SELECT hlc, model_type, record_uuid, change_type, data FROM shared_changes ORDER BY hlc ASC LIMIT ?",
-				vec![(lim as i64).into()],
+				vec![lim.into()],
 			),
 			(None, None) => Statement::from_string(
 				DbBackend::Sqlite,

core/src/service/sync/peer.rs

@@ -2715,6 +2715,12 @@ impl PeerSync {
 			"Querying shared changes from peer log"
 		);
 
+		// limit == 0 with our `len > limit` rule would return 0 rows plus
+		// has_more = true, so a naive caller would spin forever.
+		if limit == 0 {
+			return Err(anyhow::anyhow!("shared changes limit must be > 0"));
+		}
+
 		// Query peer log with SQL-side LIMIT, fetching one extra row to detect has_more
 		let fetch_limit = limit.saturating_add(1);
 		let mut entries = self

core/src/service/sync/protocol_handler.rs

@@ -114,6 +114,13 @@ impl LogSyncHandler {
 		since_hlc: Option<HLC>,
 		limit: usize,
 	) -> Result<SyncMessage> {
+		// Reject limit == 0 up front: with our `has_more = entries.len() > limit`
+		// rule it would return 0 rows with has_more = true and the caller would
+		// spin forever reissuing the same request.
+		if limit == 0 {
+			anyhow::bail!("SharedChangeRequest limit must be > 0");
+		}
+
 		// Get changes from our peer log, fetching one extra row so we can derive has_more
 		// without reloading the entire log on every batch request.
 		let fetch_limit = limit.saturating_add(1);