feat(ohos): add nearby console integration and improve UDP path selection (#2486)

* feat(ohos): complete nearby console integration

Use the core management RPC surface for ephemeral nearby deployments, harden session lifecycle and packet validation, support tunnel-to-NIC packet conversion, and timestamp HarmonyOS traffic samples.

* fix(core): prefer verified UDP hole-punch paths

Treat zero latency as unmeasured so a newly admitted UDP path cannot replace a working relay before liveness is confirmed.

---------

Co-authored-by: FrankHan <frankhan@FrankHans-Mac-mini.local>

easytier-contrib/easytier-ohrs/Cargo.lock

@@ -1244,6 +1244,8 @@ dependencies = [
  "derive_builder",
  "easytier-proto",
  "futures",
+ "getrandom 0.2.16",
+ "getrandom 0.3.3",
  "guarden",
  "hmac",
  "http-body-util",
@@ -1277,6 +1279,7 @@ dependencies = [
  "tracing",
  "url",
  "uuid",
+ "wasm-bindgen",
  "webpki-roots 0.26.11",
  "wildmatch",
  "x25519-dalek",
@@ -1288,10 +1291,15 @@ dependencies = [
 name = "easytier-ohrs"
 version = "0.1.0"
 dependencies = [
+ "anyhow",
  "async-trait",
  "base64 0.22.1",
+ "bytes",
  "easytier",
+ "easytier-core",
+ "easytier-proto",
  "flate2",
+ "futures",
  "gethostname 1.1.0",
  "ipnet",
  "napi-build-ohos",

easytier-contrib/easytier-ohrs/Cargo.toml

@@ -7,9 +7,18 @@ edition = "2024"
 crate-type=["cdylib"]
 
 [dependencies]
+anyhow = "1.0"
 async-trait = "0.1"
 base64 = "0.22"
+bytes = "1.5"
+easytier-core = { path = "../../easytier-core", default-features = false }
+easytier-proto = { path = "../../easytier-proto", default-features = false, features = [
+    "api",
+    "core",
+    "json-rpc",
+] }
 flate2 = "1.1"
+futures = "0.3"
 gethostname = "1.1"
 easytier = { path = "../../easytier" }
 napi-derive-ohos = "1.1"

easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs

@@ -20,7 +20,7 @@ use std::path::PathBuf;
 use std::sync::Mutex;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::thread::{self, JoinHandle};
-use std::time::{Duration, Instant};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 struct LocalSocketState {
     stop_flag: std::sync::Arc<AtomicBool>,
@@ -38,6 +38,7 @@ const EVENT_RECEIVER_SYNC_INTERVAL: Duration = Duration::from_secs(1);
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct TrafficStatsPayload {
+    sampled_at_ms: i64,
     instances: Vec<InstanceTrafficStats>,
 }
 
@@ -224,7 +225,7 @@ fn tun_candidate_ids(snapshot: &RuntimeAggregateState) -> HashSet<String> {
         .collect()
 }
 
-fn collect_traffic_stats() -> TrafficStatsPayload {
+fn collect_traffic_stats(sampled_at_ms: i64) -> TrafficStatsPayload {
     let running_instances = INSTANCE_MANAGER
         .instances()
         .into_iter()
@@ -285,7 +286,17 @@ fn collect_traffic_stats() -> TrafficStatsPayload {
         instances
     });
 
-    TrafficStatsPayload { instances }
+    TrafficStatsPayload {
+        sampled_at_ms,
+        instances,
+    }
+}
+
+fn unix_time_millis() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_millis().min(i64::MAX as u128) as i64)
+        .unwrap_or_default()
 }
 
 pub fn start_local_socket_server() -> bool {
@@ -409,7 +420,7 @@ pub fn start_local_socket_server() -> bool {
                 .unwrap_or(true);
             if should_collect_traffic_stats {
                 last_traffic_stats_at = Some(now);
-                match serde_json::to_string(&collect_traffic_stats()) {
+                match serde_json::to_string(&collect_traffic_stats(unix_time_millis())) {
                     Ok(json) => {
                         let _ = broadcast_local_socket_json_payload_message(
                             &mut clients,

easytier-contrib/easytier-ohrs/src/lib.rs

@@ -37,6 +37,7 @@ macro_rules! ohrs_log_debug {
 mod config;
 mod exports;
 mod kernel_bridge;
+mod nearby_management;
 mod platform;
 mod runtime;
 
@@ -57,7 +58,7 @@ use easytier::common::config::NetworkConfigExt;
 use easytier::common::constants::EASYTIER_VERSION;
 use easytier::common::{
     MachineIdOptions,
-    config::{ConfigFileControl, ConfigLoader, TomlConfigLoader},
+    config::{ConfigLoader, TomlConfigLoader},
 };
 use easytier::instance::factory::{NativeInstanceManager, native_instance_manager_with_runtime};
 use easytier::proto::api::manage::NetworkConfig;
@@ -68,6 +69,7 @@ use kernel_bridge::{
     stop_local_socket_server as stop_local_socket_server_inner,
 };
 use napi_derive_ohos::napi;
+use napi_ohos::bindgen_prelude::Uint8Array;
 use runtime::state::runtime_state::{RuntimeAggregateState, RuntimeInstanceState};
 use std::collections::{HashMap, HashSet};
 use std::format;
@@ -380,6 +382,7 @@ fn stop_runtime_inner() -> bool {
             && ok;
     }
     maybe_stop_local_socket_server();
+    let _ = nearby_management::stop_runtime_management_server();
     ok
 }
 
@@ -730,7 +733,12 @@ pub(crate) fn run_network_instance_from_json(cfg_json: &str) -> bool {
         return false;
     }
 
-    match INSTANCE_MANAGER.run_network_instance(cfg, ConfigFileControl::STATIC_CONFIG) {
+    let config_control = nearby_management::runtime_management_config_control(inst_id);
+    if !nearby_management::ensure_runtime_management_server_started() {
+        return false;
+    }
+
+    match INSTANCE_MANAGER.run_network_instance(cfg, config_control) {
         Ok(_) => {
             cache_runtime_config_snapshot(inst_id.to_string(), inst_id.to_string(), config);
             true
@@ -959,6 +967,92 @@ pub fn run_network_instance(cfg_json: String) -> bool {
     run_network_instance_from_json(&cfg_json)
 }
 
+/// Starts the management server in the VPN Extension process even when no
+/// network instance is active, allowing a nearby controller to deploy a
+/// one-shot config through the canonical Core RPC surface.
+#[napi]
+pub fn start_nearby_management_host() -> bool {
+    nearby_management::ensure_runtime_management_server_started()
+}
+
+#[napi]
+pub fn stop_nearby_management_host() -> bool {
+    nearby_management::stop_runtime_management_server()
+}
+
+#[napi]
+pub fn drain_nearby_host_commands() -> Vec<nearby_management::NearbyHostCommand> {
+    nearby_management::drain_nearby_host_commands()
+}
+
+#[napi]
+pub fn complete_nearby_host_command(
+    request_id: String,
+    success: bool,
+    error: Option<String>,
+) -> bool {
+    nearby_management::complete_nearby_host_command(request_id, success, error)
+}
+
+/// Returns 1 for canonical Core RPC packets, 2 for the OHOS-private settings
+/// envelope, and 0 for malformed or unsupported data.
+#[napi]
+pub fn nearby_management_packet_kind(packet: Uint8Array) -> i32 {
+    nearby_management::nearby_management_packet_kind(packet)
+}
+
+#[napi]
+pub fn encode_nearby_ohos_packet(envelope_json: String) -> Option<Uint8Array> {
+    nearby_management::encode_nearby_ohos_packet(envelope_json)
+}
+
+#[napi]
+pub fn decode_nearby_ohos_packet(packet: Uint8Array) -> Option<String> {
+    nearby_management::decode_nearby_ohos_packet(packet)
+}
+
+/// Opens one Core RPC endpoint for a HarmonyOS collaboration session.
+///
+/// The Harmony layer transports the returned native packets verbatim with
+/// `abilityConnectionManager.sendData`; all RPC framing stays inside Core.
+#[napi]
+pub fn open_nearby_management_session(session_key: String, host: bool) -> bool {
+    nearby_management::open_nearby_management_session(session_key, host)
+}
+
+#[napi]
+pub fn close_nearby_management_session(session_key: String) -> bool {
+    nearby_management::close_nearby_management_session(session_key)
+}
+
+#[napi]
+pub fn push_nearby_management_packet(session_key: String, packet: Uint8Array) -> bool {
+    nearby_management::push_nearby_management_packet(session_key, packet)
+}
+
+#[napi]
+pub fn drain_nearby_management_packets(session_key: String) -> Vec<Uint8Array> {
+    nearby_management::drain_nearby_management_packets(session_key)
+}
+
+#[napi]
+pub async fn call_nearby_management_json_rpc(
+    session_key: String,
+    service_name: String,
+    method_name: String,
+    domain_name: Option<String>,
+    payload_json: String,
+) -> String {
+    nearby_management::call_nearby_management_json_rpc(
+        session_key,
+        service_name,
+        method_name,
+        domain_name,
+        payload_json,
+    )
+    .await
+}
+
 #[napi]
 pub fn collect_network_infos() -> Vec<KeyValuePair> {
     exports::runtime_api::collect_network_infos()

easytier-core/src/connectivity/hole_punch/udp/connector.rs

@@ -206,7 +206,9 @@ where
                     op(true);
                     false
                 } else {
-                    tracing::info!("hole punching transport admitted successfully");
+                    tracing::info!(
+                        "hole punching transport admitted; awaiting liveness measurement"
+                    );
                     true
                 }
             }

easytier-core/src/peers/conn/peer.rs

@@ -26,6 +26,10 @@ use tokio_util::task::AbortOnDropHandle;
 type ArcPeerConn = Arc<PeerConn>;
 type ConnMap = Arc<DashMap<PeerConnId, ArcPeerConn>>;
 
+fn conn_latency_sort_key(latency_us: u64, is_hole_punched: bool) -> (bool, u64) {
+    (is_hole_punched && latency_us == 0, latency_us)
+}
+
 pub struct Peer {
     pub peer_node_id: PeerId,
     conns: ConnMap,
@@ -212,16 +216,19 @@ impl Peer {
             return Some(conn);
         }
 
-        // find a conn with the smallest latency
-        let mut min_latency = u64::MAX;
-        let mut selected = None;
-        for conn in self.conns.iter() {
-            let latency = conn.value().get_stats().latency_us;
-            if latency < min_latency {
-                min_latency = latency;
-                selected = Some(conn.value().clone());
-            }
-        }
+        // A zero latency on a hole-punched connection means the ping loop has not
+        // confirmed liveness yet. Prefer any other connection, so a freshly admitted
+        // hole-punched path cannot steal traffic before its first successful ping.
+        let selected = self
+            .conns
+            .iter()
+            .min_by_key(|conn| {
+                conn_latency_sort_key(
+                    conn.value().get_stats().latency_us,
+                    conn.value().is_hole_punched(),
+                )
+            })
+            .map(|conn| conn.value().clone());
 
         if let Some(conn) = selected.as_ref() {
             self.default_conn.store(Some(conn.clone()));
@@ -321,3 +328,23 @@ impl Drop for Peer {
         tracing::info!("peer {} drop", self.peer_node_id);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::conn_latency_sort_key;
+
+    #[test]
+    fn measured_relay_precedes_unverified_hole_punch_path() {
+        assert!(conn_latency_sort_key(20_000, false) < conn_latency_sort_key(0, true));
+    }
+
+    #[test]
+    fn unmeasured_regular_connection_keeps_existing_priority() {
+        assert!(conn_latency_sort_key(0, false) < conn_latency_sort_key(20_000, false));
+    }
+
+    #[test]
+    fn verified_lower_latency_path_can_be_preferred() {
+        assert!(conn_latency_sort_key(5_000, true) < conn_latency_sort_key(20_000, false));
+    }
+}