Fix remaining sync verification regressions

Author: lucidprogrammer

crates/contextdb-server/src/sync_client.rs

@@ -770,7 +770,7 @@ mod tests {
             let id = Uuid::new_v4();
             let mut values = HashMap::new();
             values.insert("id".to_string(), Value::Uuid(id));
-            values.insert("data".to_string(), Value::Text("x".repeat(60 * 1024)));
+            values.insert("data".to_string(), Value::Text("x".repeat(100 * 1024)));
             rows.push(RowChange {
                 table: "observations".to_string(),
                 natural_key: NaturalKey {
@@ -797,7 +797,7 @@ mod tests {
 
         assert!(
             batches.len() >= 2,
-            "10 rows * ~60KB each must split into at least 2 batches, got {}",
+            "10 rows * ~100KB each must split into at least 2 batches, got {}",
             batches.len()
         );
         let total_rows: usize = batches.iter().map(|b| b.rows.len()).sum();

crates/contextdb-server/src/sync_server.rs

@@ -69,17 +69,15 @@ impl SyncServer {
         loop {
             tokio::select! {
                 maybe_msg = push_sub.next() => {
-                    if let Some(msg) = maybe_msg {
-                        if let Err(e) = self.handle_push(&client, msg).await {
-                            tracing::error!(error = %e, "handle_push failed");
-                        }
+                    if let Some(msg) = maybe_msg
+                        && let Err(e) = self.handle_push(&client, msg).await {
+                        tracing::error!(error = %e, "handle_push failed");
                     }
                 }
                 maybe_msg = pull_sub.next() => {
-                    if let Some(msg) = maybe_msg {
-                        if let Err(e) = self.handle_pull(&client, msg).await {
-                            tracing::error!(error = %e, "handle_pull failed");
-                        }
+                    if let Some(msg) = maybe_msg
+                        && let Err(e) = self.handle_pull(&client, msg).await {
+                        tracing::error!(error = %e, "handle_pull failed");
                     }
                 }
                 _ = cleanup_interval.tick() => {

crates/contextdb-server/tests/sync_integration.rs

@@ -1042,6 +1042,7 @@ async fn a15_concurrent_push_and_pull() {
     use contextdb_core::Value;
     use uuid::Uuid;
 
+    let nats = start_nats().await;
     let server_db = Arc::new(Database::open_memory());
     let edge_db = Arc::new(Database::open_memory());
     let empty = HashMap::new();
@@ -1063,20 +1064,18 @@ async fn a15_concurrent_push_and_pull() {
         .execute("INSERT INTO t (id, v) VALUES ($id, $v)", &p)
         .unwrap();
 
-    // Register local server
     let policies = ConflictPolicies::uniform(ConflictPolicy::InsertIfNotExists);
-    let _server = SyncServer::new(
+    let server = Arc::new(SyncServer::new(
         server_db.clone(),
-        "nats://localhost:19999",
+        &nats.nats_url,
         "concurrent-client-test",
         policies.clone(),
-    );
+    ));
+    let server_handle = server.clone();
+    tokio::spawn(async move { server_handle.run().await });
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
 
-    let client = SyncClient::new(
-        edge_db.clone(),
-        "nats://localhost:19999",
-        "concurrent-client-test",
-    );
+    let client = SyncClient::new(edge_db.clone(), &nats.nats_url, "concurrent-client-test");
 
     // Insert data on edge (for push to send)
     let edge_id = Uuid::new_v4();