fix: complete parent PUT when child subscribe finishes (#2109)

Author: sanity

apps/freenet-ping/app/tests/run_app_blocked_peers.rs

@@ -831,8 +831,12 @@ async fn test_ping_blocked_peers_simple() -> TestResult {
 // as they only varied in non-functional aspects like timeouts and logging
 
 /// Solution/reference implementation for blocked peers
+// TODO-MUST-FIX: WebSocket connection reset during teardown - see issue #2108
+// Test passes functionally (PUT/GET/Subscribe/state propagation all work) but
+// fails with "Connection reset without closing handshake" during cleanup.
+// Likely a test teardown race rather than functional bug.
 #[tokio::test(flavor = "multi_thread")]
-#[ignore = "fix me"]
+#[ignore]
 async fn test_ping_blocked_peers_solution() -> TestResult {
     run_blocked_peers_test(BlockedPeersConfig {
         test_name: "solution",

crates/core/src/node/network_bridge/p2p_protoc.rs

@@ -976,6 +976,13 @@ impl P2pConnManager {
                             } => {
                                 tracing::debug!(%tx, %key, "local subscribe complete");
 
+                                // If this is a child operation, complete it and let the parent flow handle result delivery.
+                                if op_manager.is_sub_operation(tx) {
+                                    tracing::info!(%tx, %key, "completing child subscribe operation");
+                                    op_manager.completed(tx);
+                                    continue;
+                                }
+
                                 if !op_manager.is_sub_operation(tx) {
                                     let response = Ok(HostResponse::ContractResponse(
                                         ContractResponse::SubscribeResponse { key, subscribed },

crates/core/src/operations/mod.rs

@@ -126,9 +126,11 @@ where
                     "root operation awaiting child completion"
                 );
 
+                // Track the root op so child completions can finish it later.
                 op_manager
                     .root_ops_awaiting_sub_ops()
                     .insert(tx_id, final_state);
+                tracing::info!(%tx_id, "root operation registered as awaiting sub-ops");
 
                 return Ok(None);
             }

crates/core/src/ring/connection_manager.rs

@@ -423,6 +423,13 @@ impl ConnectionManager {
         self.transient_connections.contains_key(peer)
     }
 
+    #[allow(dead_code)]
+    pub fn is_transient_addr(&self, addr: &SocketAddr) -> bool {
+        self.transient_connections
+            .iter()
+            .any(|entry| entry.key().addr == *addr)
+    }
+
     pub fn transient_count(&self) -> usize {
         self.transient_in_use.load(Ordering::Acquire)
     }