perf: restore asynchronous route publication for Owner writes

fluxon_rs/fluxon_kv/src/client_kv_api/external_api.rs

@@ -468,20 +468,19 @@ async fn release_external_local_first_pending_slot(
     }
 }
 
-async fn publish_external_local_first(
+fn spawn_external_local_first_publish(
     inner: &client_kv_api::ClientKvApiInner,
     items: Vec<OwnerLocalPublishItem>,
     pending_contexts: Vec<ExternalPendingPutCtx>,
-) -> KvResult<Vec<KvResult<()>>> {
+) {
     assert_eq!(
         items.len(),
         pending_contexts.len(),
         "external local-first publish job must retain one context per item"
     );
     let view = inner.view.clone_view();
     let spawn_view = view.clone();
-    let (completion_tx, completion_rx) = ::tokio::sync::oneshot::channel();
-    // The registered owner task retains the slots and fences if the RPC is cancelled.
+    // ACK covers local commit. The owner task retains slots and fences through publication.
     spawn_view.spawn("external_local_first_route_publish", async move {
         let results = client_kv_api::put::publish_external_owner_local_job(
             view,
@@ -492,13 +491,12 @@ async fn publish_external_local_first(
             },
         )
         .await;
-        let _ = completion_tx.send(results);
+        for result in results {
+            if let Err(error) = result {
+                tracing::warn!(%error, "external local-first background publication failed");
+            }
+        }
     });
-    completion_rx.await.map_err(|_| {
-        KvError::Api(ApiError::SystemShutdown {
-            detail: "owner Put publication task stopped before completion".to_string(),
-        })
-    })
 }
 
 pub(crate) fn normalize_external_get_start_group_lens(
@@ -4854,7 +4852,6 @@ impl HandlerForExternalClient for ClientKvApi {
         let mut revoke_pending = Vec::new();
         let mut local_publish_items = Vec::new();
         let mut local_publish_contexts = Vec::new();
-        let mut local_publish_indices = Vec::new();
 
         for (idx, item) in req.items.into_iter().enumerate() {
             let Some(put_id) = item.put_id else {
@@ -4939,7 +4936,10 @@ impl HandlerForExternalClient for ClientKvApi {
                             radix: pending_ctx.radix.clone(),
                         });
                         local_publish_contexts.push(pending_ctx);
-                        local_publish_indices.push(idx);
+                        results[idx] = Some(ExternalBatchPutTransferEndItemResp {
+                            error_code: OK,
+                            error_json: String::new(),
+                        });
                     }
                     Err(err) => {
                         release_external_local_first_pending_slot(
@@ -5012,19 +5012,7 @@ impl HandlerForExternalClient for ClientKvApi {
         }
 
         if !local_publish_items.is_empty() {
-            let published =
-                publish_external_local_first(inner, local_publish_items, local_publish_contexts)
-                    .await?;
-            assert_eq!(published.len(), local_publish_indices.len());
-            for (idx, result) in local_publish_indices.into_iter().zip(published) {
-                results[idx] = Some(match result {
-                    Ok(()) => ExternalBatchPutTransferEndItemResp {
-                        error_code: OK,
-                        error_json: String::new(),
-                    },
-                    Err(error) => ExternalBatchPutTransferEndItemResp::from_error(&error),
-                });
-            }
+            spawn_external_local_first_publish(inner, local_publish_items, local_publish_contexts);
         }
 
         if !revoke_pending.is_empty() {
@@ -5269,7 +5257,6 @@ impl HandlerForExternalClient for ClientKvApi {
         let mut revoke_pending = Vec::new();
         let mut local_publish_items = Vec::new();
         let mut local_publish_contexts = Vec::new();
-        let mut local_publish_indices = Vec::new();
 
         for (idx, item) in req.items.into_iter().enumerate() {
             let Some(put_id) = item.put_id else {
@@ -5353,7 +5340,10 @@ impl HandlerForExternalClient for ClientKvApi {
                             radix: pending_ctx.radix.clone(),
                         });
                         local_publish_contexts.push(pending_ctx);
-                        local_publish_indices.push(idx);
+                        results[idx] = Some(ExternalBatchPutCommitItemResp {
+                            error_code: OK,
+                            error_json: String::new(),
+                        });
                     }
                     Err(err) => {
                         release_external_local_first_pending_slot(
@@ -5420,19 +5410,7 @@ impl HandlerForExternalClient for ClientKvApi {
         }
 
         if !local_publish_items.is_empty() {
-            let published =
-                publish_external_local_first(inner, local_publish_items, local_publish_contexts)
-                    .await?;
-            assert_eq!(published.len(), local_publish_indices.len());
-            for (idx, result) in local_publish_indices.into_iter().zip(published) {
-                results[idx] = Some(match result {
-                    Ok(()) => ExternalBatchPutCommitItemResp {
-                        error_code: OK,
-                        error_json: String::new(),
-                    },
-                    Err(error) => ExternalBatchPutCommitItemResp::from_error(&error),
-                });
-            }
+            spawn_external_local_first_publish(inner, local_publish_items, local_publish_contexts);
         }
 
         if !revoke_pending.is_empty() {

fluxon_rs/fluxon_kv/src/external_client_api/external_client_test.rs

@@ -645,15 +645,15 @@ async fn external_client_basic_crud(
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
-async fn test_external_put_completion_waits_for_route_publication() {
+async fn test_external_put_acknowledges_local_commit_before_route_publication() {
     use crate::client_kv_api::external_api::HandlerForExternalClient;
     use crate::client_kv_api::msg_pack::{
         ExternalBatchPutCommitItemReq, ExternalBatchPutCommitReq, ExternalBatchPutStartItemReq,
         ExternalBatchPutStartReq, ExternalBatchPutTransferEndItemReq,
         ExternalBatchPutTransferEndReq,
     };
     use crate::client_kv_api::{ExternalLocalFirstPutKeyReservation, ExternalPutKeyOutcome};
-    use crate::rpcresp_kvresult_convert::{msg_and_error::KvError, try_from_code};
+    use crate::rpcresp_kvresult_convert::try_from_code;
 
     #[derive(Clone, Copy, Debug)]
     enum Entry {
@@ -663,13 +663,12 @@ async fn test_external_put_completion_waits_for_route_publication() {
     #[derive(Clone, Copy, Debug, PartialEq, Eq)]
     enum Scenario {
         Success,
-        Cancel,
         RejectIndependent,
     }
 
     let _test_guard = crate::kvcore_test_lib::integration_test_lock().await;
     tokio::time::timeout(Duration::from_secs(120), async {
-        fluxon_util::test_util::start_test_etcd().expect("start etcd for publication barrier");
+        fluxon_util::test_util::start_test_etcd().expect("start etcd for local commit acknowledgement");
         let etcd = fluxon_util::dev_config::read_etcd_endpoint_from_build_config()
             .expect("read test etcd endpoint");
         let suffix = uuid::Uuid::new_v4();
@@ -696,7 +695,7 @@ async fn test_external_put_completion_waits_for_route_publication() {
         ).await;
 
         for entry in [Entry::TransferEnd, Entry::Commit] {
-            for scenario in [Scenario::Success, Scenario::Cancel, Scenario::RejectIndependent] {
+            for scenario in [Scenario::Success, Scenario::RejectIndependent] {
                 let keys = (0..3).map(|index| format!("{entry:?}-{scenario:?}-{index}"))
                     .collect::<Vec<_>>();
                 let started = owner_api.external_batch_put_start(ExternalBatchPutStartReq {
@@ -762,12 +761,18 @@ async fn test_external_put_completion_waits_for_route_publication() {
                         }
                     }
                 });
-                tokio::select! {
-                    biased;
-                    result = &mut completion => panic!("{entry:?} acknowledged before route publication: {result:?}"),
-                    reached = entered => reached.expect("publication worker reached gate"),
-                }
-                assert!(futures::poll!(&mut completion).is_pending());
+                let results = tokio::time::timeout(Duration::from_secs(5), &mut completion)
+                    .await
+                    .expect("local commit must acknowledge while Master publication is paused");
+                assert_eq!(results.len(), keys.len());
+                assert!(results.iter().all(Result::is_ok), "local commit results: {results:?}");
+                // The caller can leave after the local ACK; the owner retains publication.
+                drop(completion);
+                drop(results);
+                tokio::time::timeout(Duration::from_secs(5), entered)
+                    .await
+                    .expect("publication worker must reach the gate")
+                    .expect("publication worker reached gate");
                 for key in &keys {
                     assert!(master_inner.kv_routes.get(key).is_none(), "gate must precede the master route");
                 }
@@ -780,14 +785,15 @@ async fn test_external_put_completion_waits_for_route_publication() {
                         _ => panic!("pending publication must retain its per-key fence"),
                     }
                 }).collect::<Vec<_>>();
-                let results = if scenario == Scenario::Cancel {
-                    drop(completion);
-                    release.send(()).expect("release cancelled RPC's owner worker");
-                    None
-                } else {
-                    release.send(()).expect("release publication worker");
-                    Some(completion.await)
-                };
+                for (index, reader) in readers.iter().enumerate() {
+                    assert_eq!(reader.bytes(), vec![0x30 + index as u8; 4096]);
+                }
+                for follower in &followers {
+                    let wait = follower.wait_released();
+                    futures::pin_mut!(wait);
+                    assert!(futures::poll!(&mut wait).is_pending(), "local ACK must retain the publication fence");
+                }
+                release.send(()).expect("release the departed caller's owner publication");
                 for (index, follower) in followers.into_iter().enumerate() {
                     let rejected = scenario == Scenario::RejectIndependent && index == 0;
                     assert_eq!(follower.wait().await, if rejected {
@@ -798,14 +804,11 @@ async fn test_external_put_completion_waits_for_route_publication() {
                     follower.wait_released().await;
                     assert_eq!(readers[index].bytes(), vec![0x30 + index as u8; 4096]);
                     if rejected {
-                        assert!(matches!(results.as_ref().unwrap()[index], Err(KvError::LeaseMgr(_))));
+                        // Publication rejection does not retract the local ACK or invalidate holders.
                         assert!(master_inner.kv_routes.get(&keys[index]).is_none());
                         assert!(inner.local_visible_mem_holder(&keys[index]).is_none());
                     } else {
-                        if let Some(results) = &results {
-                            assert!(results[index].is_ok(), "{:?}", results[index]);
-                        }
-                        let route = master_inner.kv_routes.get(&keys[index]).expect("successful Put must publish its route");
+                        let route = master_inner.kv_routes.get(&keys[index]).expect("background publication must eventually expose the route");
                         assert_eq!(route.atomic_group.as_ref().map(|group| group.members.len()), (index > 0).then_some(2));
                         drop(route);
                         let (holder, _) = inner.get(&keys[index]).await.expect("get completed value").expect("published value exists");
@@ -815,12 +818,31 @@ async fn test_external_put_completion_waits_for_route_publication() {
                 let after = inner.runtime_observe_snapshot();
                 assert!(after.grouped_put_done_batches > before.grouped_put_done_batches);
                 assert_eq!(after.legacy_put_done_batches, before.legacy_put_done_batches);
+                assert_eq!(
+                    after.owner_segment_allocated_bytes,
+                    before.owner_segment_allocated_bytes,
+                    "a rejected publication must retain its physical slot while a reader holds it"
+                );
+                let rejected_capacity = if scenario == Scenario::RejectIndependent {
+                    readers[0].local_reserve_resident_slot_ref()
+                        .expect("rejected local value retains its exact slot").2
+                } else {
+                    0
+                };
+                let released = inner.owner_local_reserve_rebalance_notify();
                 drop(readers);
+                tokio::time::timeout(
+                    Duration::from_secs(5),
+                    fluxon_util::notify_state::wait_until(&released, || {
+                        inner.owner_local_reserve_control_snapshot().allocated_bytes
+                            == after.owner_segment_allocated_bytes - rejected_capacity
+                    }),
+                ).await.expect("the last rejected reader must release its slot without freeing published values");
             }
         }
         owner.shutdown().await.expect("stop publication owner");
         master.shutdown().await.expect("stop publication master");
-    }).await.expect("external publication barriers and fence cleanup must complete");
+    }).await.expect("local ACK, background publication, and fence cleanup must complete");
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
@@ -1247,6 +1269,7 @@ async fn test_external_remote_get_uses_bound_owner_slot_without_master_allocator
         )
         .await
         .expect("put only on source owner");
+    wait_master_route(master_view.clone(), key, Duration::from_secs(5)).await;
     {
         let route = master_view
             .master_kv_router()

fluxon_rs/fluxon_kv/src/kv_test.rs

@@ -302,6 +302,25 @@ async fn get_owner_probe_value(
     }
 }
 
+async fn wait_for_external_route_publication(
+    view: &crate::external_client_api::ExternalClientApiView,
+    key: &str,
+    deadline: Instant,
+) -> anyhow::Result<()> {
+    let api = view.external_client_api();
+    timeout(deadline.saturating_duration_since(Instant::now()), async {
+        loop {
+            // Default existence checks query Master instead of the local snapshot.
+            if api.inner().is_exist(key).await? {
+                return Ok::<(), anyhow::Error>(());
+            }
+            sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .map_err(|_| anyhow::anyhow!("external route publication timed out: key={}", key))?
+}
+
 async fn fetch_transfer_link_te_value(
     cluster_name: &str,
     from_instance_key: &str,
@@ -381,6 +400,20 @@ async fn verify_rdma_transfer_data_link(
             continue;
         }
 
+        // The probe reads from another Owner, so it needs the Master route.
+        wait_for_external_route_publication(
+            &probe_writer_framework.external_client_api_view(),
+            attempt_key.as_str(),
+            deadline,
+        )
+        .await
+        .unwrap_or_else(|err| {
+            panic!(
+                "closed transfer data probe publication failed: cluster={} from={} to={} key={} attempt={} err={}",
+                cluster_name, from_instance_key, to_instance_key, attempt_key, attempt, err
+            )
+        });
+
         let payload = match get_owner_probe_value(requester_framework, attempt_key.as_str()).await {
             Ok(payload) => payload,
             Err(err) => {
@@ -1302,6 +1335,14 @@ async fn perform_external_client_crud_operations(
         }
     }
 
+    // Same-owner GET above checks local completion; global existence needs publication.
+    wait_for_external_route_publication(
+        &view,
+        key,
+        Instant::now() + Duration::from_secs(KV_TEST_TRANSFER_PROBE_IO_TIMEOUT_SECS),
+    )
+    .await?;
+
     // EXIST操作
     match api.inner().is_exist(key).await {
         Ok(true) => info!(