perf: eliminate full message-history deep clone in add_messages reducer

Cargo.toml

@@ -70,6 +70,7 @@ dotenvy = "0.15.7"
 uuid = { workspace = true }
 
 [dev-dependencies]
+mimalloc = "0.1"
 langgraph = { path = ".", features = ["full"] }
 langgraph-prebuilt = { workspace = true }
 langgraph-providers = { workspace = true }

crates/langgraph-checkpoint-sqlite/src/saver.rs

@@ -341,14 +341,13 @@ impl SqliteSaver {
         let mut results = Vec::with_capacity(rows.len());
         for row in rows {
             let mut tuple = Self::row_to_tuple(&row)?;
-            // Reconcile channel values from blobs.
+            // Merge blob values (version-joined) over the row's channel_values
+            // (see aget_tuple for the rationale).
             let thread_id = row.get::<String, _>("thread_id");
             let ns = row.get::<String, _>("checkpoint_ns");
             let cid = tuple.checkpoint.id.clone();
             let blob_values = self.load_blobs(&thread_id, &ns, &cid).await?;
-            if !blob_values.is_empty() {
-                tuple.checkpoint.channel_values = blob_values;
-            }
+            tuple.checkpoint.channel_values.extend(blob_values);
             tuple.pending_writes = Some(self.load_writes(&thread_id, &ns, &cid).await?);
             results.push(tuple);
         }
@@ -517,10 +516,12 @@ impl BaseCheckpointSaver for SqliteSaver {
 
         let mut tuple = Self::row_to_tuple(&row)?;
         let cid = tuple.checkpoint.id.clone();
+        // Blob rows are version-addressed and written incrementally, so the
+        // join in SELECT_BLOBS_SQL resolves every channel's value regardless
+        // of which checkpoint created the blob. Merge (not replace) so values
+        // still resolve even if a blob row is ever missing.
         let blob_values = self.load_blobs(thread_id, checkpoint_ns, &cid).await?;
-        if !blob_values.is_empty() {
-            tuple.checkpoint.channel_values = blob_values;
-        }
+        tuple.checkpoint.channel_values.extend(blob_values);
         tuple.pending_writes = Some(self.load_writes(thread_id, checkpoint_ns, &cid).await?);
         Ok(Some(tuple))
     }
@@ -548,8 +549,11 @@ impl BaseCheckpointSaver for SqliteSaver {
 
         let next_config = Self::make_config(thread_id, checkpoint_ns, &checkpoint.id);
 
-        // Strip channel_values from the JSON checkpoint payload to avoid
-        // duplicating them in the row body — they live in checkpoint_blobs.
+        // Strip channel_values from the JSON checkpoint payload — they live in
+        // checkpoint_blobs. Blob rows are keyed by (channel, version) and are
+        // written incrementally (only channels in `new_versions`), so on load
+        // every channel's value resolves through the version join regardless
+        // of which checkpoint wrote the blob.
         let mut checkpoint_value = serde_json::to_value(checkpoint)
             .map_err(|e| CheckpointError::Storage(e.to_string()))?;
         if let Some(obj) = checkpoint_value.as_object_mut() {
@@ -967,6 +971,85 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_incremental_blob_writes() {
+        // cp1: channels a and b both written at version 1.
+        let saver = fresh_saver().await;
+        let cfg = config_for("thread-INC");
+        let mut cp1 = Checkpoint::empty();
+        cp1.channel_values
+            .insert("a".into(), JsonValue::Number(1.into()));
+        cp1.channel_values
+            .insert("b".into(), JsonValue::Number(1.into()));
+        cp1.channel_versions
+            .insert("a".into(), JsonValue::String("1".into()));
+        cp1.channel_versions
+            .insert("b".into(), JsonValue::String("1".into()));
+        let mut versions1: ChannelVersions = HashMap::new();
+        versions1.insert("a".into(), JsonValue::String("1".into()));
+        versions1.insert("b".into(), JsonValue::String("1".into()));
+        let next1 = saver
+            .aput(&cfg, &cp1, &CheckpointMetadata::default(), &versions1)
+            .await
+            .unwrap();
+
+        // cp2: only channel a is updated — new_versions carries just the delta.
+        let mut cp2 = Checkpoint::empty();
+        cp2.channel_values
+            .insert("a".into(), JsonValue::Number(2.into()));
+        cp2.channel_values
+            .insert("b".into(), JsonValue::Number(1.into()));
+        cp2.channel_versions
+            .insert("a".into(), JsonValue::String("2".into()));
+        cp2.channel_versions
+            .insert("b".into(), JsonValue::String("1".into()));
+        let mut versions2: ChannelVersions = HashMap::new();
+        versions2.insert("a".into(), JsonValue::String("2".into()));
+        saver
+            .aput(&next1, &cp2, &CheckpointMetadata::default(), &versions2)
+            .await
+            .unwrap();
+
+        // Blob rows are keyed by (channel, version): cp2 only adds a@2, so
+        // the table holds exactly a@1, b@1, a@2 (a full rewrite would add b@2).
+        let blob_count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM checkpoint_blobs WHERE thread_id = 'thread-INC'",
+        )
+        .fetch_one(&saver.pool)
+        .await
+        .unwrap();
+        assert_eq!(
+            blob_count, 3,
+            "expected blobs a@1, b@1, a@2, got {blob_count}"
+        );
+
+        // cp2 reads back both channels: a@2 via cp2's blob, b@1 via the
+        // version-addressed blob written by cp1 (b's version never moved).
+        let cfg_cp2 = config_with_id("thread-INC", &cp2.id);
+        let tuple = saver.aget_tuple(&cfg_cp2).await.unwrap().unwrap();
+        assert_eq!(
+            tuple.checkpoint.channel_values.get("a"),
+            Some(&JsonValue::Number(2.into()))
+        );
+        assert_eq!(
+            tuple.checkpoint.channel_values.get("b"),
+            Some(&JsonValue::Number(1.into()))
+        );
+
+        // cp1 remains fully readable (older rows rely on blobs, and b@1 still
+        // matches cp2's join as well since b's version never moved).
+        let cfg_cp1 = config_with_id("thread-INC", &cp1.id);
+        let earlier = saver.aget_tuple(&cfg_cp1).await.unwrap().unwrap();
+        assert_eq!(
+            earlier.checkpoint.channel_values.get("a"),
+            Some(&JsonValue::Number(1.into()))
+        );
+        assert_eq!(
+            earlier.checkpoint.channel_values.get("b"),
+            Some(&JsonValue::Number(1.into()))
+        );
+    }
+
     #[tokio::test]
     async fn test_metadata_filter_returns_only_matching_rows() {
         let saver = fresh_saver().await;

crates/langgraph-core/src/channels/binop.rs

@@ -4,7 +4,10 @@ use parking_lot::RwLock;
 use serde_json::Value as JsonValue;
 
 /// Reducer function type: (current, update) -> new
-pub type ReducerFn = fn(&JsonValue, &JsonValue) -> JsonValue;
+///
+/// Takes ownership of `current` so reducers can move existing state instead of
+/// deep-cloning it on every update; `update` is borrowed (usually small).
+pub type ReducerFn = fn(JsonValue, &JsonValue) -> JsonValue;
 
 /// Applies a binary operator to accumulate values.
 ///
@@ -67,7 +70,9 @@ impl Channel for BinaryOperatorAggregate {
                 continue;
             }
 
-            match guard.as_ref() {
+            // Move the accumulated value out so the reducer can consume it
+            // instead of deep-cloning the whole state on every update.
+            match guard.take() {
                 Some(current) => {
                     let new_val = (self.reducer)(current, val);
                     *guard = Some(new_val);
@@ -102,10 +107,10 @@ impl Channel for BinaryOperatorAggregate {
 }
 
 /// Common reducer: append arrays
-pub fn append_reducer(current: &JsonValue, update: &JsonValue) -> JsonValue {
+pub fn append_reducer(current: JsonValue, update: &JsonValue) -> JsonValue {
     let mut result = match current {
-        JsonValue::Array(arr) => arr.clone(),
-        other => vec![other.clone()],
+        JsonValue::Array(arr) => arr,
+        other => vec![other],
     };
     match update {
         JsonValue::Array(arr) => result.extend(arr.iter().cloned()),
@@ -115,10 +120,10 @@ pub fn append_reducer(current: &JsonValue, update: &JsonValue) -> JsonValue {
 }
 
 /// Common reducer: merge objects
-pub fn merge_reducer(current: &JsonValue, update: &JsonValue) -> JsonValue {
+pub fn merge_reducer(current: JsonValue, update: &JsonValue) -> JsonValue {
     match (current, update) {
         (JsonValue::Object(curr), JsonValue::Object(upd)) => {
-            let mut merged = curr.clone();
+            let mut merged = curr;
             for (k, v) in upd {
                 merged.insert(k.clone(), v.clone());
             }

crates/langgraph-core/src/graph/state.rs

@@ -482,13 +482,19 @@ impl CompiledStateGraph {
     }
 
     /// Save a checkpoint from current channel state.
+    ///
+    /// `previous_versions` holds the channel versions as of the start of this
+    /// run; only channels whose version moved since then are passed to the
+    /// saver as `new_versions`, so persistent savers can write delta blobs
+    /// instead of re-encoding every channel on every super-step.
     fn save_checkpoint(
         &self,
         checkpointer: &Arc<dyn BaseCheckpointSaver>,
         config: &RunnableConfig,
         channels: &HashMap<String, Box<dyn Channel>>,
         channel_versions: &ChannelVersions,
         versions_seen: &HashMap<String, HashMap<String, JsonValue>>,
+        previous_versions: &ChannelVersions,
     ) -> Option<RunnableConfig> {
         use chrono::Utc;
         use langgraph_checkpoint::checkpoint::id::uuid6;
@@ -509,9 +515,17 @@ impl CompiledStateGraph {
             updated_channels: None,
         };
 
+        // Delta vs. the versions this run started from: only channels whose
+        // version changed since then need new blob rows.
+        let new_versions: ChannelVersions = channel_versions
+            .iter()
+            .filter(|(k, v)| previous_versions.get(k.as_str()) != Some(*v))
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect();
+
         let metadata = CheckpointMetadata::default();
         checkpointer
-            .put(config, &checkpoint, &metadata, channel_versions)
+            .put(config, &checkpoint, &metadata, &new_versions)
             .ok()
     }
 
@@ -789,6 +803,7 @@ impl CompiledStateGraph {
             .as_ref()
             .map(|s| s.checkpoint.versions_seen.clone())
             .unwrap_or_default();
+        let previous_versions = channel_versions.clone();
 
         // Apply the update values to channels
         if let Some(obj) = values.as_object() {
@@ -817,6 +832,7 @@ impl CompiledStateGraph {
             &channels,
             &channel_versions,
             &versions_seen,
+            &previous_versions,
         );
 
         Ok(config.clone())
@@ -1405,6 +1421,10 @@ impl CompiledStateGraph {
                 )
             };
 
+        // Baseline for incremental checkpoint writes: the versions as of the
+        // start of this run. Only channels that move past these get blob rows.
+        let previous_versions = channel_versions.clone();
+
         // BSP loop counters
         let mut step: u64 = 0;
         let max_steps = config.get_recursion_limit().unwrap_or(self.recursion_limit);
@@ -1528,6 +1548,7 @@ impl CompiledStateGraph {
                             &channels,
                             &channel_versions,
                             &versions_seen,
+                            &previous_versions,
                         ) {
                             config = new_config;
                         }
@@ -1600,6 +1621,7 @@ impl CompiledStateGraph {
                             &channels,
                             &channel_versions,
                             &versions_seen,
+                            &previous_versions,
                         ) {
                             config = new_config;
                         }
@@ -1683,9 +1705,14 @@ impl CompiledStateGraph {
 
             // Save "loop" checkpoint after each completed super-step
             if let Some(ref cp) = self.checkpointer {
-                if let Some(new_config) =
-                    self.save_checkpoint(cp, &config, &channels, &channel_versions, &versions_seen)
-                {
+                if let Some(new_config) = self.save_checkpoint(
+                    cp,
+                    &config,
+                    &channels,
+                    &channel_versions,
+                    &versions_seen,
+                    &previous_versions,
+                ) {
                     config = new_config;
                 }
             }