perf: optimize store batch serialization, zero-clone semantic search, and lock contention

server/src/store.rs

@@ -444,14 +444,35 @@ impl<T: DeserializeOwned + Default + Serialize + Send + Sync + 'static> Store<T>
             (*lock).clone()
         };
 
-        // Expensive serialization and granular extraction run completely unblocked outside the lock
-        let full_bytes_res = serde_json::to_vec(&new_snapshot);
-        let granular_entries = full_bytes_res
-            .as_ref()
-            .ok()
-            .and_then(|bytes| serde_json::from_slice::<serde_json::Value>(bytes).ok())
-            .map(|val| Self::extract_granular_entries(&self.key, &val))
-            .unwrap_or_default();
+        match self.prepare_batch(&new_snapshot) {
+            Ok((batch_inserts, removed_keys)) => {
+                self.queue.push_batch(
+                    self.key.clone(),
+                    batch_inserts,
+                    removed_keys,
+                    self.flushed.clone(),
+                );
+            }
+            Err(e) => tracing::error!(
+                "Failed to serialize memory store for key '{}': {}",
+                self.key,
+                e
+            ),
+        }
+    }
+
+    fn prepare_batch(
+        &self,
+        new_snapshot: &T,
+    ) -> Result<(Vec<(String, Vec<u8>)>, Vec<String>), serde_json::Error>
+    where
+        T: Serialize,
+    {
+        let full_bytes = serde_json::to_vec(new_snapshot)?;
+        let granular_entries = match serde_json::from_slice::<serde_json::Value>(&full_bytes) {
+            Ok(val) => Self::extract_granular_entries(&self.key, &val),
+            Err(_) => Vec::new(),
+        };
 
         let new_keys: std::collections::HashSet<String> =
             granular_entries.iter().map(|(k, _)| k.clone()).collect();
@@ -469,48 +490,11 @@ impl<T: DeserializeOwned + Default + Serialize + Send + Sync + 'static> Store<T>
             *known = new_keys;
         }
 
-        match full_bytes_res {
-            Ok(data) => {
-                let mut batch_inserts = Vec::with_capacity(granular_entries.len() + 1);
-                for (g_key, g_bytes) in granular_entries {
-                    batch_inserts.push((g_key, g_bytes));
-                }
-                batch_inserts.push((self.key.clone(), data.clone()));
+        let mut batch_inserts = Vec::with_capacity(granular_entries.len() + 1);
+        batch_inserts.extend(granular_entries);
+        batch_inserts.push((self.key.clone(), full_bytes));
 
-                if self
-                    .queue
-                    .push_batch(
-                        self.key.clone(),
-                        batch_inserts.clone(),
-                        removed_keys.clone(),
-                        self.flushed.clone(),
-                    )
-                    .is_none()
-                {
-                    tracing::warn!(
-                        "DbWriteQueue channel full for key '{}'. Applying backpressure fallback.",
-                        self.key
-                    );
-                    let queue = self.queue.clone();
-                    let key = self.key.clone();
-                    let flushed = self.flushed.clone();
-                    if let Ok(handle) = tokio::runtime::Handle::try_current() {
-                        handle.spawn(async move {
-                            let _ = tokio::time::timeout(
-                                std::time::Duration::from_secs(10),
-                                queue.push_batch_async(key, batch_inserts, removed_keys, flushed),
-                            )
-                            .await;
-                        });
-                    }
-                }
-            }
-            Err(e) => tracing::error!(
-                "Failed to serialize memory store for key '{}': {}",
-                self.key,
-                e
-            ),
-        }
+        Ok((batch_inserts, removed_keys))
     }
 
     pub async fn modify_async<F: FnOnce(&mut T)>(&self, f: F)
@@ -531,38 +515,8 @@ impl<T: DeserializeOwned + Default + Serialize + Send + Sync + 'static> Store<T>
             (*lock).clone()
         };
 
-        let full_bytes_res = serde_json::to_vec(&new_snapshot);
-        let granular_entries = full_bytes_res
-            .as_ref()
-            .ok()
-            .and_then(|bytes| serde_json::from_slice::<serde_json::Value>(bytes).ok())
-            .map(|val| Self::extract_granular_entries(&self.key, &val))
-            .unwrap_or_default();
-
-        let new_keys: std::collections::HashSet<String> =
-            granular_entries.iter().map(|(k, _)| k.clone()).collect();
-        let mut removed_keys = Vec::new();
-        {
-            let mut known = self
-                .known_granular_keys
-                .write()
-                .unwrap_or_else(|e| e.into_inner());
-            for old_k in known.iter() {
-                if !new_keys.contains(old_k) {
-                    removed_keys.push(old_k.clone());
-                }
-            }
-            *known = new_keys;
-        }
-
-        match full_bytes_res {
-            Ok(data) => {
-                let mut batch_inserts = Vec::with_capacity(granular_entries.len() + 1);
-                for (g_key, g_bytes) in granular_entries {
-                    batch_inserts.push((g_key, g_bytes));
-                }
-                batch_inserts.push((self.key.clone(), data));
-
+        match self.prepare_batch(&new_snapshot) {
+            Ok((batch_inserts, removed_keys)) => {
                 if let Some(rx) = self
                     .queue
                     .push_batch_async(