fix: ensure originating peer updates local cache on subsequent PUTs

When a contract is already seeded locally, subsequent PUT operations
were skipping local state updates with "skipping duplicate caching".
This caused the originating peer to serve stale cached state even after
successfully PUTting updated state to the network.

Now the originating peer always calls put_contract() to upsert its
local state, regardless of whether the contract is already seeded.
The seed_contract() call is only skipped if already seeded.

Also adds integration test `test_put_merge_persists_state()` that:
- PUTs initial state
- PUTs updated state to same contract
- Verifies both originating peer and gateway serve updated state

This test would have caught both:
- Issue #1995 (target peer not persisting merged state)
- This issue (originating peer not updating local cache)

Related to #1995

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: sanity

crates/core/src/operations/put.rs

@@ -547,40 +547,34 @@ impl Operation for PutOp {
                             // Check if already stored before any operations
                             let is_seeding_contract = op_manager.ring.is_seeding_contract(&key);
 
-                            // Only store the contract locally if not already seeded
-                            if !is_seeding_contract {
-                                tracing::debug!(
-                                    tx = %id,
-                                    %key,
-                                    peer = %op_manager.ring.connection_manager.get_peer_key().unwrap(),
-                                    "Storing contract locally after successful put"
-                                );
+                            // Always store/update the contract locally
+                            tracing::debug!(
+                                tx = %id,
+                                %key,
+                                peer = %op_manager.ring.connection_manager.get_peer_key().unwrap(),
+                                is_seeding = is_seeding_contract,
+                                "Storing contract locally after successful put"
+                            );
 
-                                // Store the contract locally
-                                put_contract(
-                                    op_manager,
-                                    key,
-                                    state.clone(),
-                                    RelatedContracts::default(),
-                                    &contract.clone(),
-                                )
-                                .await?;
+                            // Store the contract locally (will upsert if already exists)
+                            put_contract(
+                                op_manager,
+                                key,
+                                state.clone(),
+                                RelatedContracts::default(),
+                                &contract.clone(),
+                            )
+                            .await?;
 
-                                // Always seed the contract locally after a successful put
+                            // Mark as seeded if not already seeded
+                            if !is_seeding_contract {
                                 tracing::debug!(
                                     tx = %id,
                                     %key,
                                     peer = %op_manager.ring.connection_manager.get_peer_key().unwrap(),
                                     "Adding contract to local seed list"
                                 );
                                 op_manager.ring.seed_contract(key);
-                            } else {
-                                tracing::debug!(
-                                    tx = %id,
-                                    %key,
-                                    peer = %op_manager.ring.connection_manager.get_peer_key().unwrap(),
-                                    "Contract already seeded locally, skipping duplicate caching"
-                                );
                             }
 
                             // Start subscription if the contract is already seeded and the user requested it

crates/core/tests/operations.rs

@@ -537,6 +537,279 @@ async fn test_update_contract() -> TestResult {
     Ok(())
 }
 
+/// Test that a second PUT to an already cached contract persists the merged state.
+/// This is a regression test for issue #1995.
+#[test_log::test(tokio::test(flavor = "multi_thread", worker_threads = 4))]
+async fn test_put_merge_persists_state() -> TestResult {
+    // Load test contract
+    const TEST_CONTRACT: &str = "test-contract-integration";
+    let contract = test_utils::load_contract(TEST_CONTRACT, vec![].into())?;
+    let contract_key = contract.key();
+
+    // Create initial state with empty todo list
+    let initial_state = test_utils::create_empty_todo_list();
+    let initial_wrapped_state = WrappedState::from(initial_state);
+
+    // Create network sockets
+    let network_socket_b = TcpListener::bind("127.0.0.1:0")?;
+    let ws_api_port_socket_a = TcpListener::bind("127.0.0.1:0")?;
+    let ws_api_port_socket_b = TcpListener::bind("127.0.0.1:0")?;
+
+    // Configure gateway node B
+    let (config_b, preset_cfg_b, config_b_gw) = {
+        let (cfg, preset) = base_node_test_config(
+            true,
+            vec![],
+            Some(network_socket_b.local_addr()?.port()),
+            ws_api_port_socket_b.local_addr()?.port(),
+        )
+        .await?;
+        let public_port = cfg.network_api.public_port.unwrap();
+        let path = preset.temp_dir.path().to_path_buf();
+        (cfg, preset, gw_config(public_port, &path)?)
+    };
+    let ws_api_port_peer_b = config_b.ws_api.ws_api_port.unwrap();
+
+    // Configure peer node A
+    let (config_a, preset_cfg_a) = base_node_test_config(
+        false,
+        vec![serde_json::to_string(&config_b_gw)?],
+        None,
+        ws_api_port_socket_a.local_addr()?.port(),
+    )
+    .await?;
+    let ws_api_port_peer_a = config_a.ws_api.ws_api_port.unwrap();
+
+    tracing::info!("Node A data dir: {:?}", preset_cfg_a.temp_dir.path());
+    tracing::info!("Node B (gw) data dir: {:?}", preset_cfg_b.temp_dir.path());
+
+    // Start node A (peer)
+    std::mem::drop(ws_api_port_socket_a);
+    let node_a = async move {
+        let _span = with_peer_id("peer-a");
+        tracing::info!("Starting peer A node");
+        let config = config_a.build().await?;
+        let node = NodeConfig::new(config.clone())
+            .await?
+            .build(serve_gateway(config.ws_api).await)
+            .await?;
+        tracing::info!("Peer A node running");
+        node.run().await
+    }
+    .boxed_local();
+
+    // Start node B (gateway)
+    std::mem::drop(network_socket_b);
+    std::mem::drop(ws_api_port_socket_b);
+    let node_b = async {
+        let _span = with_peer_id("gateway");
+        tracing::info!("Starting gateway node");
+        let config = config_b.build().await?;
+        let node = NodeConfig::new(config.clone())
+            .await?
+            .build(serve_gateway(config.ws_api).await)
+            .await?;
+        tracing::info!("Gateway node running");
+        node.run().await
+    }
+    .boxed_local();
+
+    let test = tokio::time::timeout(Duration::from_secs(180), async {
+        // Wait for nodes to start up
+        tracing::info!("Waiting for nodes to start up...");
+        tokio::time::sleep(Duration::from_secs(15)).await;
+        tracing::info!("Nodes should be ready, proceeding with test...");
+
+        // Connect to node A's websocket API
+        let uri = format!(
+            "ws://127.0.0.1:{ws_api_port_peer_a}/v1/contract/command?encodingProtocol=native"
+        );
+        let (stream, _) = connect_async(&uri).await?;
+        let mut client_api_a = WebApi::start(stream);
+
+        // First PUT: Store initial contract state
+        tracing::info!("Sending first PUT with initial state...");
+        make_put(
+            &mut client_api_a,
+            initial_wrapped_state.clone(),
+            contract.clone(),
+            false,
+        )
+        .await?;
+
+        // Wait for first put response
+        let resp = tokio::time::timeout(Duration::from_secs(120), client_api_a.recv()).await;
+        match resp {
+            Ok(Ok(HostResponse::ContractResponse(ContractResponse::PutResponse { key }))) => {
+                tracing::info!("First PUT successful for contract: {}", key);
+                assert_eq!(key, contract_key);
+            }
+            Ok(Ok(other)) => {
+                bail!("Unexpected response for first PUT: {:?}", other);
+            }
+            Ok(Err(e)) => {
+                bail!("Error receiving first PUT response: {}", e);
+            }
+            Err(_) => {
+                bail!("Timeout waiting for first PUT response");
+            }
+        }
+
+        // Wait a bit to ensure state is fully cached
+        tokio::time::sleep(Duration::from_secs(2)).await;
+
+        // Create updated state with more data (simulating a state merge)
+        let mut updated_todo_list: test_utils::TodoList =
+            serde_json::from_slice(initial_wrapped_state.as_ref()).unwrap();
+
+        // Add multiple tasks to make the state larger
+        for i in 1..=5 {
+            updated_todo_list.tasks.push(test_utils::Task {
+                id: i,
+                title: format!("Task {}", i),
+                description: format!("Description for task {}", i),
+                completed: false,
+                priority: i as u8,
+            });
+        }
+
+        let updated_bytes = serde_json::to_vec(&updated_todo_list).unwrap();
+        let updated_wrapped_state = WrappedState::from(updated_bytes);
+
+        tracing::info!(
+            "Initial state size: {} bytes, Updated state size: {} bytes",
+            initial_wrapped_state.as_ref().len(),
+            updated_wrapped_state.as_ref().len()
+        );
+
+        // Second PUT: Update the already-cached contract with new state
+        // This tests the bug fix - the merged state should be persisted
+        tracing::info!("Sending second PUT with updated state...");
+        make_put(
+            &mut client_api_a,
+            updated_wrapped_state.clone(),
+            contract.clone(),
+            false,
+        )
+        .await?;
+
+        // Wait for second put response
+        let resp = tokio::time::timeout(Duration::from_secs(120), client_api_a.recv()).await;
+        match resp {
+            Ok(Ok(HostResponse::ContractResponse(ContractResponse::PutResponse { key }))) => {
+                tracing::info!("Second PUT successful for contract: {}", key);
+                assert_eq!(key, contract_key);
+            }
+            Ok(Ok(other)) => {
+                bail!("Unexpected response for second PUT: {:?}", other);
+            }
+            Ok(Err(e)) => {
+                bail!("Error receiving second PUT response: {}", e);
+            }
+            Err(_) => {
+                bail!("Timeout waiting for second PUT response");
+            }
+        }
+
+        // Wait a bit to ensure the merge and persistence completes
+        tokio::time::sleep(Duration::from_secs(2)).await;
+
+        // Now GET the contract from both nodes and verify they both return the updated state
+        tracing::info!("Getting contract from peer A to verify updated state persisted...");
+        let (response_contract_a, response_state_a) =
+            get_contract(&mut client_api_a, contract_key, &preset_cfg_a.temp_dir).await?;
+
+        assert_eq!(response_contract_a.key(), contract_key);
+
+        // Deserialize and verify the state from peer A
+        let response_todo_list_a: test_utils::TodoList =
+            serde_json::from_slice(response_state_a.as_ref())
+                .expect("Failed to deserialize state from peer A");
+
+        tracing::info!(
+            "Peer A returned state with {} tasks, size {} bytes",
+            response_todo_list_a.tasks.len(),
+            response_state_a.as_ref().len()
+        );
+
+        // This is the key assertion: the state should have the 5 tasks from the second PUT
+        // If the bug exists, it would return the empty todo list from the first PUT
+        assert_eq!(
+            response_todo_list_a.tasks.len(),
+            5,
+            "Peer A should return updated state with 5 tasks, not cached stale state"
+        );
+
+        // Verify the state size matches (another way to catch the bug)
+        assert_eq!(
+            response_state_a.as_ref().len(),
+            updated_wrapped_state.as_ref().len(),
+            "State size should match the updated state, not the initial cached state"
+        );
+
+        // Also verify from gateway node B
+        let uri = format!(
+            "ws://127.0.0.1:{ws_api_port_peer_b}/v1/contract/command?encodingProtocol=native"
+        );
+        let (stream, _) = connect_async(&uri).await?;
+        let mut client_api_b = WebApi::start(stream);
+
+        tracing::info!("Getting contract from gateway to verify state propagated...");
+        let (response_contract_b, response_state_b) =
+            get_contract(&mut client_api_b, contract_key, &preset_cfg_b.temp_dir).await?;
+
+        assert_eq!(response_contract_b.key(), contract_key);
+
+        let response_todo_list_b: test_utils::TodoList =
+            serde_json::from_slice(response_state_b.as_ref())
+                .expect("Failed to deserialize state from gateway");
+
+        tracing::info!(
+            "Gateway returned state with {} tasks, size {} bytes",
+            response_todo_list_b.tasks.len(),
+            response_state_b.as_ref().len()
+        );
+
+        assert_eq!(
+            response_todo_list_b.tasks.len(),
+            5,
+            "Gateway should also return updated state with 5 tasks"
+        );
+
+        tracing::info!(
+            "✓ Test passed: Second PUT correctly persisted merged state (issue #1995 fixed)"
+        );
+
+        // Cleanup
+        client_api_a
+            .send(ClientRequest::Disconnect { cause: None })
+            .await?;
+        client_api_b
+            .send(ClientRequest::Disconnect { cause: None })
+            .await?;
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        Ok::<_, anyhow::Error>(())
+    });
+
+    select! {
+        a = node_a => {
+            let Err(a) = a;
+            return Err(anyhow!(a).into());
+        }
+        b = node_b => {
+            let Err(b) = b;
+            return Err(anyhow!(b).into());
+        }
+        r = test => {
+            r??;
+            tokio::time::sleep(Duration::from_secs(3)).await;
+        }
+    }
+
+    Ok(())
+}
+
 // This test is disabled due to race conditions in subscription propagation logic.
 // The test expects multiple clients across different nodes to receive subscription updates,
 // but the PUT caching refactor (commits 2cd337b5-0d432347) changed the subscription semantics.