diff --git a/docs/AndroidDevelopment.md b/docs/AndroidDevelopment.md index 8720b561..8f172347 100644 --- a/docs/AndroidDevelopment.md +++ b/docs/AndroidDevelopment.md @@ -509,6 +509,6 @@ Tauri 版本升级后也必须重新验证运行时 envelope，不能仅凭 API ## 12. LAN Sync 多播发现 -Android 多播发现需要 `CHANGE_WIFI_MULTICAST_STATE` 权限。宿主只在前台持有 `MulticastLock`，进入后台或停止发现时释放；网络逻辑由 Rust 负责。 +Android 的 Rust mDNS 发现需要 `CHANGE_WIFI_MULTICAST_STATE` 权限。宿主只在前台持有 `MulticastLock`，进入后台或停止发现时释放，回到前台后为已启动的发现重新获取；网络逻辑由 Rust 负责。 平台适配见 [LanDiscoveryPlugin](../src-tauri/crates/tauritavern/gen/android/app/src/main/java/com/tauritavern/client/LanDiscoveryPlugin.kt)，功能边界见[同步总览](CurrentState/Sync.md)。 diff --git a/docs/CurrentState/Sync.md b/docs/CurrentState/Sync.md index a1001ebd..d923de87 100644 --- a/docs/CurrentState/Sync.md +++ b/docs/CurrentState/Sync.md @@ -32,6 +32,8 @@ LAN 上传保留发起方的覆盖策略，但实际回拉使用目标设备的 本机广播随接收服务启停，只下载时可以独立发现。发现失败不删除配对或停用仍可用的 HTTPS 连接；停止接收服务不取消已接受的本地下载作业。 +附近发现使用 mDNS / DNS-SD。iOS 与 macOS 使用系统 Bonjour，Android、Windows 与 Linux 使用 Rust mDNS；两套后端共享服务信息与设备目录，记录更新和过期由发现后端处理。 + LAN 功能受宿主 `sync.lan` 能力门禁和平台权限约束。移动端权限、签名与生命周期要求见 [Android](../AndroidDevelopment.md) 和 [iOS](../iOSDevelopment.md)。 ## 面板、事件与自动同步 diff --git a/docs/iOSDevelopment.md b/docs/iOSDevelopment.md index 7db4875e..f5442545 100644 --- a/docs/iOSDevelopment.md +++ b/docs/iOSDevelopment.md @@ -275,8 +275,8 @@ iOS Chat Completion 的后台租约由 Rust `ChatCompletionService` 持有，不 当前不使用 `BGAppRefreshTask` / `BGProcessingTask`，因为它们由系统择机调度，不能表达用户点击后立即开始的聊天生成。`BGContinuedProcessingTask` 只拥有系统后台执行权和展示状态；真正的网络请求、取消和流会话生命周期仍由既有 Rust 服务负责，不引入第二套生成调度状态机。 -## 8. LAN Sync 多播发现权限 +## 8. LAN Sync Bonjour 发现 -LAN 发现需要本地网络用途声明和 [Multicast Networking entitlement](https://developer.apple.com/documentation/bundleresources/entitlements/com.apple.developer.networking.multicast)。签名使用的 App ID 与 provisioning profile 必须获得相应授权并包含该 entitlement；仓库声明不能代替 Apple 授权。 +LAN 发现通过系统 Bonjour 浏览和注册 `_tauritavern._tcp`。Info.plist 声明 `NSLocalNetworkUsageDescription` 与 `NSBonjourServices`，由用户授予本地网络访问权限；此路径不需要 Multicast Networking entitlement。macOS 使用相同的系统后端与用途声明。 权限行为以真机签名包验证，后台发现受 iOS 生命周期限制。配置入口见 [Apple host 工程](../src-tauri/crates/tauritavern/gen/apple/project.yml)，功能边界见[同步总览](CurrentState/Sync.md)。 diff --git a/packaging/flatpak/cargo-sources.json b/packaging/flatpak/cargo-sources.json index 58e700b8..8950966d 100644 --- a/packaging/flatpak/cargo-sources.json +++ b/packaging/flatpak/cargo-sources.json @@ -207,6 +207,19 @@ "dest": "cargo/vendor/async-compression-0.4.43", "dest-filename": ".cargo-checksum.json" }, + { + "type": "archive", + "archive-type": "tar-gzip", + "url": "https://static.crates.io/crates/async-dnssd/async-dnssd-0.5.1.crate", + "sha256": "3d49ffe175ab45bbfd74b548313d9d7cdfff27161a94b007b52eeeb5f9aaa15e", + "dest": "cargo/vendor/async-dnssd-0.5.1" + }, + { + "type": "inline", + "contents": "{\"package\": \"3d49ffe175ab45bbfd74b548313d9d7cdfff27161a94b007b52eeeb5f9aaa15e\", \"files\": {}}", + "dest": "cargo/vendor/async-dnssd-0.5.1", + "dest-filename": ".cargo-checksum.json" + }, { "type": "archive", "archive-type": "tar-gzip", @@ -2521,6 +2534,19 @@ "dest": "cargo/vendor/fluent-uri-0.4.1", "dest-filename": ".cargo-checksum.json" }, + { + "type": "archive", + "archive-type": "tar-gzip", + "url": "https://static.crates.io/crates/flume/flume-0.12.0.crate", + "sha256": "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be", + "dest": "cargo/vendor/flume-0.12.0" + }, + { + "type": "inline", + "contents": "{\"package\": \"5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be\", \"files\": {}}", + "dest": "cargo/vendor/flume-0.12.0", + "dest-filename": ".cargo-checksum.json" + }, { "type": "archive", "archive-type": "tar-gzip", @@ -4406,6 +4432,19 @@ "dest": "cargo/vendor/idna_adapter-1.2.2", "dest-filename": ".cargo-checksum.json" }, + { + "type": "archive", + "archive-type": "tar-gzip", + "url": "https://static.crates.io/crates/if-addrs/if-addrs-0.15.0.crate", + "sha256": "c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b", + "dest": "cargo/vendor/if-addrs-0.15.0" + }, + { + "type": "inline", + "contents": "{\"package\": \"c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b\", \"files\": {}}", + "dest": "cargo/vendor/if-addrs-0.15.0", + "dest-filename": ".cargo-checksum.json" + }, { "type": "archive", "archive-type": "tar-gzip", @@ -5212,6 +5251,19 @@ "dest": "cargo/vendor/matchit-0.8.4", "dest-filename": ".cargo-checksum.json" }, + { + "type": "archive", + "archive-type": "tar-gzip", + "url": "https://static.crates.io/crates/mdns-sd/mdns-sd-0.21.3.crate", + "sha256": "a63c9b854b5ad0812ac5969f8db92a59f28b45d0a5ae599b9a45b84c5c8a08e8", + "dest": "cargo/vendor/mdns-sd-0.21.3" + }, + { + "type": "inline", + "contents": "{\"package\": \"a63c9b854b5ad0812ac5969f8db92a59f28b45d0a5ae599b9a45b84c5c8a08e8\", \"files\": {}}", + "dest": "cargo/vendor/mdns-sd-0.21.3", + "dest-filename": ".cargo-checksum.json" + }, { "type": "archive", "archive-type": "tar-gzip", @@ -6369,6 +6421,19 @@ "dest": "cargo/vendor/pin-project-lite-0.2.17", "dest-filename": ".cargo-checksum.json" }, + { + "type": "archive", + "archive-type": "tar-gzip", + "url": "https://static.crates.io/crates/pin-utils/pin-utils-0.1.0.crate", + "sha256": "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184", + "dest": "cargo/vendor/pin-utils-0.1.0" + }, + { + "type": "inline", + "contents": "{\"package\": \"8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184\", \"files\": {}}", + "dest": "cargo/vendor/pin-utils-0.1.0", + "dest-filename": ".cargo-checksum.json" + }, { "type": "archive", "archive-type": "tar-gzip", @@ -8150,6 +8215,19 @@ "dest": "cargo/vendor/smallvec-1.15.2", "dest-filename": ".cargo-checksum.json" }, + { + "type": "archive", + "archive-type": "tar-gzip", + "url": "https://static.crates.io/crates/socket-pktinfo/socket-pktinfo-0.4.1.crate", + "sha256": "612942246d0cc239cfd83af1dfd39be47f649208a3524e5e9da651910128e0ac", + "dest": "cargo/vendor/socket-pktinfo-0.4.1" + }, + { + "type": "inline", + "contents": "{\"package\": \"612942246d0cc239cfd83af1dfd39be47f649208a3524e5e9da651910128e0ac\", \"files\": {}}", + "dest": "cargo/vendor/socket-pktinfo-0.4.1", + "dest-filename": ".cargo-checksum.json" + }, { "type": "archive", "archive-type": "tar-gzip", @@ -8215,6 +8293,19 @@ "dest": "cargo/vendor/soup3-sys-0.5.0", "dest-filename": ".cargo-checksum.json" }, + { + "type": "archive", + "archive-type": "tar-gzip", + "url": "https://static.crates.io/crates/spin/spin-0.9.9.crate", + "sha256": "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e", + "dest": "cargo/vendor/spin-0.9.9" + }, + { + "type": "inline", + "contents": "{\"package\": \"3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e\", \"files\": {}}", + "dest": "cargo/vendor/spin-0.9.9", + "dest-filename": ".cargo-checksum.json" + }, { "type": "archive", "archive-type": "tar-gzip", diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index f4138edc..c37efac0 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -172,6 +172,25 @@ dependencies = [ "tokio", ] +[[package]] +name = "async-dnssd" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d49ffe175ab45bbfd74b548313d9d7cdfff27161a94b007b52eeeb5f9aaa15e" +dependencies = [ + "bitflags 1.3.2", + "futures-channel", + "futures-core", + "futures-executor", + "futures-util", + "libc", + "log", + "pin-utils", + "pkg-config", + "tokio", + "winapi", +] + [[package]] name = "async-executor" version = "1.14.0" @@ -1974,6 +1993,17 @@ dependencies = [ "serde", ] +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -3884,6 +3914,16 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "if-addrs" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "image" version = "0.25.10" @@ -4541,6 +4581,21 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "mdns-sd" +version = "0.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a63c9b854b5ad0812ac5969f8db92a59f28b45d0a5ae599b9a45b84c5c8a08e8" +dependencies = [ + "fastrand", + "flume", + "if-addrs", + "log", + "mio", + "socket-pktinfo", + "socket2", +] + [[package]] name = "memchr" version = "2.8.3" @@ -4644,6 +4699,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -5435,6 +5491,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "piper" version = "0.2.5" @@ -6920,6 +6982,17 @@ version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +[[package]] +name = "socket-pktinfo" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "612942246d0cc239cfd83af1dfd39be47f649208a3524e5e9da651910128e0ac" +dependencies = [ + "libc", + "socket2", + "windows-sys 0.61.2", +] + [[package]] name = "socket2" version = "0.6.5" @@ -6989,6 +7062,15 @@ dependencies = [ "system-deps", ] +[[package]] +name = "spin" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" +dependencies = [ + "lock_api", +] + [[package]] name = "spki" version = "0.7.3" @@ -8459,12 +8541,15 @@ dependencies = [ name = "tt-adapter-sync" version = "0.1.0" dependencies = [ + "async-dnssd", "async-trait", "axum", "axum-server", "base64 0.23.1", "filetime", + "futures-util", "local-ip-address", + "mdns-sd", "rand 0.10.2", "reqwest", "rustls", diff --git a/src-tauri/crates/tauritavern/Info.macos.plist b/src-tauri/crates/tauritavern/Info.macos.plist index 92752296..653b8866 100644 --- a/src-tauri/crates/tauritavern/Info.macos.plist +++ b/src-tauri/crates/tauritavern/Info.macos.plist @@ -2,6 +2,10 @@ + NSBonjourServices + + _tauritavern._tcp + NSLocalNetworkUsageDescription TauriTavern uses Local Network access to discover and sync with your paired devices on the same network. diff --git a/src-tauri/crates/tauritavern/gen/apple/project.yml b/src-tauri/crates/tauritavern/gen/apple/project.yml index d475f38b..1ea1b598 100644 --- a/src-tauri/crates/tauritavern/gen/apple/project.yml +++ b/src-tauri/crates/tauritavern/gen/apple/project.yml @@ -39,6 +39,7 @@ targets: path: tauritavern_iOS/Info.plist properties: LSRequiresIPhoneOS: true + NSBonjourServices: ["_tauritavern._tcp"] UILaunchStoryboardName: LaunchScreen UIRequiredDeviceCapabilities: [arm64, metal] UISupportedInterfaceOrientations: @@ -54,8 +55,6 @@ targets: CFBundleVersion: "1.0.0" entitlements: path: tauritavern_iOS/tauritavern_iOS.entitlements - properties: - com.apple.developer.networking.multicast: true scheme: environmentVariables: RUST_BACKTRACE: full diff --git a/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/Info.plist b/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/Info.plist index 513a99a5..a8c0cc60 100644 --- a/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/Info.plist +++ b/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/Info.plist @@ -30,6 +30,10 @@ TauriTavern uses the camera only to scan LAN Sync pairing QR codes between your devices. NSLocalNetworkUsageDescription TauriTavern uses Local Network access only for LAN Sync between your devices on the same Wi‑Fi. + NSBonjourServices + + _tauritavern._tcp + NSPhotoLibraryAddUsageDescription TauriTavern saves images to your photo library only when you choose to save an image. LSRequiresIPhoneOS diff --git a/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/tauritavern_iOS.entitlements b/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/tauritavern_iOS.entitlements index ec706aa6..0c67376e 100644 --- a/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/tauritavern_iOS.entitlements +++ b/src-tauri/crates/tauritavern/gen/apple/tauritavern_iOS/tauritavern_iOS.entitlements @@ -1,8 +1,5 @@ - - com.apple.developer.networking.multicast - - + diff --git a/src-tauri/crates/tauritavern/src/app/host/shutdown.rs b/src-tauri/crates/tauritavern/src/app/host/shutdown.rs index 7b811aa4..c8290cb0 100644 --- a/src-tauri/crates/tauritavern/src/app/host/shutdown.rs +++ b/src-tauri/crates/tauritavern/src/app/host/shutdown.rs @@ -58,12 +58,7 @@ pub(super) fn handle_run_event(app_handle: &tauri::AppHandle, event: tauri::RunE { let service = state.services.lan_sync_service.clone(); tauri::async_runtime::spawn(async move { - if service - .get_status() - .await - .is_ok_and(|status| status.running) - && let Err(error) = service.discover_devices().await - { + if let Err(error) = service.refresh_discovery_if_started().await { tracing::warn!("Failed to refresh LAN discovery after resume: {error}"); } }); diff --git a/src-tauri/crates/tt-adapter-sync/Cargo.toml b/src-tauri/crates/tt-adapter-sync/Cargo.toml index 7c87d7e7..ef936ce3 100644 --- a/src-tauri/crates/tt-adapter-sync/Cargo.toml +++ b/src-tauri/crates/tt-adapter-sync/Cargo.toml @@ -15,7 +15,6 @@ reqwest = { workspace = true, features = ["json", "rustls", "stream"] } rustls = { version = "0.23", default-features = false, features = ["aws_lc_rs", "std", "tls12"] } serde = { workspace = true } serde_json = { workspace = true } -socket2 = { version = "0.6", features = ["all"] } tokio = { version = "1", default-features = false, features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "sync", "time"] } tracing = "0.1" tt-contracts = { path = "../tt-contracts" } @@ -28,5 +27,16 @@ ttsync-http = { workspace = true } url = { workspace = true } uuid = { workspace = true } +[target.'cfg(any(target_os = "macos", target_os = "ios"))'.dependencies] +async-dnssd = "0.5.1" +futures-util = "0.3" + +[target.'cfg(any(target_os = "windows", target_os = "linux", target_os = "android"))'.dependencies] +mdns-sd = "0.21.3" + [dev-dependencies] base64 = "0.23" +socket2 = { version = "0.6", features = ["all"] } + +[target.'cfg(target_os = "macos")'.dev-dependencies] +mdns-sd = "0.21.3" diff --git a/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery.rs b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery.rs index 911fdd5c..98331aa2 100644 --- a/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery.rs +++ b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery.rs @@ -1,13 +1,9 @@ use std::collections::HashMap; -use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::net::Ipv4Addr; use std::sync::{Arc, Mutex, RwLock, Weak}; -use std::time::{Duration, Instant}; +use std::time::Duration; use async_trait::async_trait; -use local_ip_address::list_afinet_netifas; -use serde::{Deserialize, Serialize}; -use socket2::{Domain, Protocol, Socket, Type}; -use tokio::net::UdpSocket; use tokio::sync::Mutex as AsyncMutex; use tokio::task::JoinHandle; use tt_contracts::lan_discovery::{LanDiscoveredDevice, LanDiscoveryAnnouncement}; @@ -16,14 +12,16 @@ use tt_domain::models::lan_sync::validate_device_name; use tt_ports::lan_discovery::{LanDeviceDiscovery, LanDiscoveryHost}; use ttsync_contract::peer::DeviceId; -const DISCOVERY_PORT: u16 = 53318; -const MULTICAST_GROUP: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 168); -const DISCOVERY_PROTOCOL: &str = "tauritavern-lan-v2"; -const MAX_PACKET_BYTES: usize = 2048; +#[cfg(any(target_os = "macos", target_os = "ios"))] +#[path = "peer_discovery/apple.rs"] +mod backend; +#[cfg(any(target_os = "windows", target_os = "linux", target_os = "android"))] +#[path = "peer_discovery/mdns.rs"] +mod backend; + +const SERVICE_TYPE: &str = "_tauritavern._tcp"; const MAX_DEVICES: usize = 128; const MAX_ADDRESSES_PER_DEVICE: usize = 8; -const REFRESH_INTERVAL: Duration = Duration::from_secs(30); -const ADDRESS_TTL: Duration = Duration::from_secs(90); const DISCOVERY_WINDOW: Duration = Duration::from_millis(500); #[derive(Clone)] @@ -39,41 +37,29 @@ struct DiscoveryInner { } struct DiscoveryRuntime { - sockets: Vec>, - receivers: Vec>, - refresher: Option>, + backend: backend::Backend, + receiver: JoinHandle>, } impl DiscoveryRuntime { - async fn stop_receivers(&mut self) { - for receiver in &self.receivers { - receiver.abort(); + async fn stop(mut self) -> Result<(), DomainError> { + self.receiver.abort(); + let received = (&mut self.receiver).await; + let stopped = self.backend.stop().await; + match received { + Ok(result) => result.and(stopped), + Err(error) if error.is_cancelled() => stopped, + Err(error) => Err(discovery_error(error)), } - for receiver in self.receivers.drain(..) { - let _ = receiver.await; - } - self.sockets.clear(); } } impl Drop for DiscoveryRuntime { fn drop(&mut self) { - for receiver in &self.receivers { - receiver.abort(); - } - if let Some(refresher) = &self.refresher { - refresher.abort(); - } + self.receiver.abort(); } } -#[derive(Serialize, Deserialize)] -struct DiscoveryMessage { - protocol: String, - request: bool, - device: Option, -} - impl LanPeerDiscovery { pub fn new(host: Arc) -> Self { Self { @@ -101,13 +87,11 @@ impl LanPeerDiscovery { } pub fn snapshot(&self) -> Vec { - let mut peers = self.inner.peers.lock().expect("LAN discovery peer lock"); - let changed = peers.expire(Instant::now()); - let devices = peers.snapshot(); - if changed { - self.inner.host.publish_changed(devices.clone()); - } - devices + self.inner + .peers + .lock() + .expect("LAN discovery peer lock") + .snapshot() } pub fn candidate_urls(&self, device_id: &DeviceId) -> Vec { @@ -120,81 +104,72 @@ impl LanPeerDiscovery { pub async fn stop(&self) -> Result<(), DomainError> { let mut runtime = self.inner.runtime.lock().await; - if let Some(mut running) = runtime.take() { - if let Some(refresher) = running.refresher.take() { - refresher.abort(); - let _ = refresher.await; - } - running.stop_receivers().await; - } + let stopped = match runtime.take() { + Some(running) => running.stop().await, + None => Ok(()), + }; *self .inner .local_device .write() .expect("LAN discovery local device lock") = None; - let had_peers = { - let mut peers = self.inner.peers.lock().expect("LAN discovery peer lock"); - let had_peers = !peers.devices.is_empty(); - peers.devices.clear(); - had_peers - }; - if had_peers { - self.inner.host.publish_changed(Vec::new()); - } - self.inner.host.set_enabled(false).await + self.inner.clear_peers(); + let released = self.inner.host.set_enabled(false).await; + stopped.and(released) } - /// Re-enumerate interfaces so foreground resume and Wi-Fi changes replace stale sockets. pub async fn refresh(&self) -> Result<(), DomainError> { - let mut runtime = self.inner.runtime.lock().await; - // Android can release its multicast lock while Rust sockets survive suspension. - self.inner.host.set_enabled(true).await?; - let running = runtime.get_or_insert_with(|| DiscoveryRuntime { - sockets: Vec::new(), - receivers: Vec::new(), - // Keep retrying when discovery was enabled before Wi-Fi became available. - refresher: Some(spawn_refresh(Arc::downgrade(&self.inner))), - }); - let sockets = match bind_sockets() { - Ok(sockets) => sockets, - Err(error) => { - if running.sockets.is_empty() { - self.inner - .host - .set_enabled(false) - .await - .map_err(|cleanup| { - DomainError::InternalError(format!( - "{error}; discovery cleanup failed: {cleanup}" - )) - })?; - } - return Err(error); - } - }; - running.stop_receivers().await; - running.sockets = sockets; - running.receivers = receive_tasks(&running.sockets, &self.inner); + self.refresh_runtime(false).await + } - self.snapshot(); - let message = DiscoveryMessage { - protocol: DISCOVERY_PROTOCOL.to_string(), - request: true, - device: self - .inner - .local_device - .read() - .expect("LAN discovery local device lock") - .clone(), - }; - let payload = serde_json::to_vec(&message) - .map_err(|error| DomainError::InvalidData(error.to_string()))?; - send_multicast(&running.sockets, &payload).await + async fn refresh_runtime(&self, only_if_started: bool) -> Result<(), DomainError> { + let mut runtime = self.inner.runtime.lock().await; + if only_if_started && runtime.is_none() { + return Ok(()); + } + if runtime + .as_ref() + .is_some_and(|running| running.receiver.is_finished()) + { + if let Err(error) = runtime + .take() + .expect("finished discovery runtime") + .stop() + .await + { + tracing::warn!("Restarting LAN discovery after failure: {error}"); + } + self.inner.clear_peers(); + } + if runtime.is_none() { + let (backend, receiver) = backend::start(Arc::downgrade(&self.inner))?; + *runtime = Some(DiscoveryRuntime { backend, receiver }); + } + self.inner.host.set_enabled(true).await?; + let local = self + .inner + .local_device + .read() + .expect("LAN discovery local device lock") + .clone(); + if let Some(device) = local { + runtime + .as_mut() + .expect("started discovery runtime") + .backend + .announce(&device) + .await?; + } + Ok(()) } } #[async_trait] impl LanDeviceDiscovery for LanPeerDiscovery { + async fn refresh_if_started(&self) -> Result<(), DomainError> { + self.refresh_runtime(true).await + } + async fn set_device_name(&self, name: &str) -> Result<(), DomainError> { validate_device_name(name)?; let announcing = { @@ -217,102 +192,116 @@ impl LanDeviceDiscovery for LanPeerDiscovery { async fn discover_devices(&self) -> Result, DomainError> { self.refresh().await?; tokio::time::sleep(DISCOVERY_WINDOW).await; + let mut runtime = self.inner.runtime.lock().await; + if runtime + .as_ref() + .is_some_and(|running| running.receiver.is_finished()) + { + let result = runtime + .take() + .expect("finished discovery runtime") + .stop() + .await; + self.inner.clear_peers(); + let released = self.inner.host.set_enabled(false).await; + result.and(released)?; + return Err(discovery_error( + "Device discovery stopped; refresh to try again", + )); + } Ok(self.snapshot()) } } -fn spawn_refresh(inner: Weak) -> JoinHandle<()> { - tokio::spawn(async move { - loop { - tokio::time::sleep(REFRESH_INTERVAL).await; - let Some(inner) = inner.upgrade() else { return }; - if let Err(error) = (LanPeerDiscovery { inner }).refresh().await { - tracing::warn!("LAN discovery refresh failed: {error}"); - } - } - }) -} - -fn receive_tasks(sockets: &[Arc], inner: &Arc) -> Vec> { - sockets - .iter() - .map(|socket| tokio::spawn(receive_loop(socket.clone(), Arc::downgrade(inner)))) - .collect() -} - -async fn receive_loop(socket: Arc, inner: Weak) { - // Windows reports an oversized receive as a socket error instead of truncating it. - // Read one full datagram, then enforce the smaller discovery message limit. - let mut buffer = vec![0; 65_536]; - loop { - let (size, source) = match socket.recv_from(&mut buffer).await { - Ok(received) => received, - Err(error) => { - tracing::warn!("LAN discovery receive failed: {error}"); - return; - } - }; - let Some(inner) = inner.upgrade() else { return }; - let Some(response) = process_message(&inner, &buffer[..size], source.ip(), Instant::now()) - else { - continue; - }; - let payload = serde_json::to_vec(&response).expect("discovery message serializes"); - if let Err(error) = socket.send_to(&payload, source).await { - tracing::warn!("LAN discovery reply to {source} failed: {error}"); +impl DiscoveryInner { + fn clear_peers(&self) { + let mut peers = self.peers.lock().expect("LAN discovery peer lock"); + if !peers.devices.is_empty() { + peers.devices.clear(); + self.host.publish_changed(Vec::new()); } } } -fn process_message( - inner: &DiscoveryInner, - bytes: &[u8], - source: IpAddr, - now: Instant, -) -> Option { - let IpAddr::V4(source) = source else { - return None; - }; - if bytes.len() > MAX_PACKET_BYTES - || source.is_unspecified() - || source.is_multicast() - || source.is_broadcast() - { - return None; - } - let message: DiscoveryMessage = serde_json::from_slice(bytes).ok()?; - if message.protocol != DISCOVERY_PROTOCOL { - return None; - } - let local = inner +fn update_peer( + inner: &Weak, + source: &str, + device: Option<(LanDiscoveryAnnouncement, Vec)>, +) { + let Some(inner) = inner.upgrade() else { return }; + let local_id = inner .local_device .read() .expect("LAN discovery local device lock") - .clone(); - if let Some(device) = message.device { - validate_announcement(&device).ok()?; - if local - .as_ref() - .is_some_and(|local| local.device_id == device.device_id) - { - return None; - } - let mut peers = inner.peers.lock().expect("LAN discovery peer lock"); - if peers.record(device, source, now) { - // Serialize snapshots with mutations so concurrent interfaces cannot publish - // an older directory after a newer one. - inner.host.publish_changed(peers.snapshot()); - } + .as_ref() + .map(|device| device.device_id.clone()); + let mut peers = inner.peers.lock().expect("LAN discovery peer lock"); + let previous = peers.snapshot(); + peers.remove_source(source); + if let Some((device, addresses)) = device + && local_id.as_ref() != Some(&device.device_id) + { + peers.record(source, device, addresses); } - if message.request { - local.map(|device| DiscoveryMessage { - protocol: DISCOVERY_PROTOCOL.to_string(), - request: false, - device: Some(device), + let devices = peers.snapshot(); + if devices != previous { + // Serialize publication with mutations, including callbacks from different interfaces. + inner.host.publish_changed(devices); + } +} + +fn discovery_error(error: impl std::fmt::Display) -> DomainError { + DomainError::Transient(format!("LAN device discovery failed: {error}")) +} + +fn txt_properties(device: &LanDiscoveryAnnouncement) -> Vec<(&'static str, Vec)> { + // Two fixed halves preserve the existing 64-character name limit within TXT's 255-byte entries. + let mut name = device.device_name.chars(); + let first: String = name.by_ref().take(32).collect(); + let second: String = name.collect(); + vec![ + ("v", b"2".to_vec()), + ("id", device.device_id.as_str().as_bytes().to_vec()), + ("name", first.into_bytes()), + ("name2", second.into_bytes()), + ( + "platform", + device + .platform + .as_deref() + .unwrap_or_default() + .as_bytes() + .to_vec(), + ), + ("spki", device.spki_sha256.as_bytes().to_vec()), + ] +} + +fn decode_device<'a>( + port: u16, + property: impl Fn(&str) -> Option<&'a [u8]>, +) -> Result { + let text = |key: &str| { + std::str::from_utf8(property(key).unwrap_or_default()).map_err(|error| { + DomainError::InvalidData(format!("Invalid LAN discovery {key}: {error}")) }) - } else { - None + }; + if text("v")? != "2" { + return Err(DomainError::InvalidData( + "Unsupported LAN discovery version".to_string(), + )); } + let platform = text("platform")?; + let device = LanDiscoveryAnnouncement { + device_id: DeviceId::new(text("id")?.to_string()) + .map_err(|error| DomainError::InvalidData(error.to_string()))?, + device_name: format!("{}{}", text("name")?, text("name2")?), + platform: (!platform.is_empty()).then(|| platform.to_string()), + port, + spki_sha256: text("spki")?.to_string(), + }; + validate_announcement(&device)?; + Ok(device) } pub(crate) fn validate_announcement(device: &LanDiscoveryAnnouncement) -> Result<(), DomainError> { @@ -335,77 +324,6 @@ pub(crate) fn validate_announcement(device: &LanDiscoveryAnnouncement) -> Result Ok(()) } -fn bind_sockets() -> Result>, DomainError> { - let interfaces = list_afinet_netifas().map_err(|error| { - DomainError::InternalError(format!("Cannot list LAN discovery interfaces: {error}")) - })?; - let mut addresses = interfaces - .into_iter() - .filter_map(|(_, ip)| match ip { - IpAddr::V4(ip) if !ip.is_loopback() && !ip.is_unspecified() && !ip.is_multicast() => { - Some(ip) - } - _ => None, - }) - .collect::>(); - addresses.sort_unstable(); - addresses.dedup(); - let mut sockets = Vec::new(); - let mut last_error = None; - for address in addresses { - match bind_socket(address) { - Ok(socket) => sockets.push(Arc::new(socket)), - Err(error) => { - tracing::warn!("Cannot join LAN discovery multicast on {address}: {error}"); - last_error = Some(format!( - "Cannot join LAN discovery multicast on {address}: {error}" - )); - } - } - } - if sockets.is_empty() { - return Err(DomainError::InternalError(last_error.unwrap_or_else( - || "No IPv4 interface available for LAN discovery".to_string(), - ))); - } - Ok(sockets) -} - -fn bind_socket(interface: Ipv4Addr) -> std::io::Result { - let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?; - socket.set_reuse_address(true)?; - #[cfg(unix)] - socket.set_reuse_port(true)?; - // Wildcard binding receives multicast on macOS; outgoing interface selection avoids - // announcing every address through the default route on multihomed devices. - socket.bind(&SocketAddr::from((Ipv4Addr::UNSPECIFIED, DISCOVERY_PORT)).into())?; - socket.join_multicast_v4(&MULTICAST_GROUP, &interface)?; - socket.set_multicast_if_v4(&interface)?; - socket.set_multicast_loop_v4(true)?; - socket.set_multicast_ttl_v4(1)?; - socket.set_nonblocking(true)?; - UdpSocket::from_std(socket.into()) -} - -async fn send_multicast(sockets: &[Arc], payload: &[u8]) -> Result<(), DomainError> { - let mut sent = false; - for socket in sockets { - match socket - .send_to(payload, (MULTICAST_GROUP, DISCOVERY_PORT)) - .await - { - Ok(_) => sent = true, - Err(error) => tracing::warn!("LAN discovery announcement failed: {error}"), - } - } - if !sent { - return Err(DomainError::InternalError( - "LAN discovery announcement failed on every interface".to_string(), - )); - } - Ok(()) -} - #[derive(Default)] struct PeerDirectory { devices: HashMap, @@ -415,57 +333,45 @@ struct SeenDevice { device_name: String, platform: Option, spki_sha256: String, - addresses: HashMap, + sources: HashMap>, } impl PeerDirectory { - fn record(&mut self, device: LanDiscoveryAnnouncement, source: Ipv4Addr, now: Instant) -> bool { - let mut changed = self.expire(now); - if self.devices.len() == MAX_DEVICES && !self.devices.contains_key(&device.device_id) { - return changed; + fn record(&mut self, source: &str, device: LanDiscoveryAnnouncement, addresses: Vec) { + let mut base_urls = addresses + .into_iter() + .filter(|ip| !ip.is_unspecified() && !ip.is_multicast() && !ip.is_broadcast()) + .map(|ip| format!("https://{ip}:{}", device.port)) + .collect::>(); + base_urls.sort_unstable(); + base_urls.dedup(); + base_urls.truncate(MAX_ADDRESSES_PER_DEVICE); + if base_urls.is_empty() + || (self.devices.len() == MAX_DEVICES && !self.devices.contains_key(&device.device_id)) + { + return; } - let peer = self.devices.entry(device.device_id).or_insert_with(|| { - changed = true; - SeenDevice { + let peer = self + .devices + .entry(device.device_id) + .or_insert_with(|| SeenDevice { device_name: device.device_name.clone(), platform: device.platform.clone(), spki_sha256: device.spki_sha256.clone(), - addresses: HashMap::new(), - } - }); - changed |= peer.device_name != device.device_name - || peer.spki_sha256 != device.spki_sha256 - || peer.platform != device.platform; + sources: HashMap::new(), + }); peer.device_name = device.device_name; peer.platform = device.platform; peer.spki_sha256 = device.spki_sha256; - let address = format!("https://{source}:{}", device.port); - if !peer.addresses.contains_key(&address) - && peer.addresses.len() == MAX_ADDRESSES_PER_DEVICE - { - let oldest = peer - .addresses - .iter() - .min_by_key(|(_, seen)| **seen) - .map(|(address, _)| address.clone()) - .expect("nonempty address cache"); - peer.addresses.remove(&oldest); - } - changed |= peer.addresses.insert(address, now).is_none(); - changed + peer.sources.insert(source.to_string(), base_urls); } - fn expire(&mut self, now: Instant) -> bool { - let mut changed = false; + fn remove_source(&mut self, source: &str) { + // ponytail: scan at most 128 peers instead of maintaining a second source-to-device index. self.devices.retain(|_, peer| { - peer.addresses.retain(|_, seen| { - let fresh = now.duration_since(*seen) < ADDRESS_TTL; - changed |= !fresh; - fresh - }); - !peer.addresses.is_empty() + peer.sources.remove(source); + !peer.sources.is_empty() }); - changed } fn snapshot(&self) -> Vec { @@ -473,8 +379,10 @@ impl PeerDirectory { .devices .iter() .map(|(device_id, peer)| { - let mut base_urls = peer.addresses.keys().cloned().collect::>(); + let mut base_urls = peer.sources.values().flatten().cloned().collect::>(); base_urls.sort_unstable(); + base_urls.dedup(); + base_urls.truncate(MAX_ADDRESSES_PER_DEVICE); LanDiscoveredDevice { device_id: device_id.clone(), device_name: peer.device_name.clone(), @@ -490,88 +398,4 @@ impl PeerDirectory { } #[cfg(test)] -mod tests { - use super::*; - - #[derive(Default)] - struct RecordingHost { - snapshots: Mutex>>, - } - - #[async_trait] - impl LanDiscoveryHost for RecordingHost { - async fn set_enabled(&self, _enabled: bool) -> Result<(), DomainError> { - Ok(()) - } - - fn publish_changed(&self, devices: Vec) { - self.snapshots.lock().unwrap().push(devices); - } - } - - #[test] - fn discovery_ignores_bad_datagrams_and_merges_source_addresses_until_they_expire() { - let host = Arc::new(RecordingHost::default()); - let discovery = LanPeerDiscovery::new(host.clone()); - let device_id = DeviceId::new("11111111-1111-4111-8111-111111111111".to_string()).unwrap(); - let packet = serde_json::to_vec(&DiscoveryMessage { - protocol: DISCOVERY_PROTOCOL.to_string(), - request: false, - device: Some(LanDiscoveryAnnouncement { - device_id: device_id.clone(), - device_name: "Peer".to_string(), - platform: Some("android".to_string()), - port: 56000, - spki_sha256: "a".repeat(43), - }), - }) - .unwrap(); - let now = Instant::now(); - let first = "192.168.1.10".parse().unwrap(); - let second = "192.168.1.20".parse().unwrap(); - process_message(&discovery.inner, b"not JSON", first, now); - let mut oversized = packet.clone(); - oversized.resize(MAX_PACKET_BYTES + 1, b' '); - process_message(&discovery.inner, &oversized, first, now); - assert!(discovery.snapshot().is_empty()); - - process_message(&discovery.inner, &packet, first, now); - process_message(&discovery.inner, &packet, first, now); - assert_eq!(host.snapshots.lock().unwrap().len(), 1); - process_message( - &discovery.inner, - &packet, - second, - now + Duration::from_secs(30), - ); - assert_eq!( - discovery.candidate_urls(&device_id), - ["https://192.168.1.10:56000", "https://192.168.1.20:56000"] - ); - - let mut renamed: DiscoveryMessage = serde_json::from_slice(&packet).unwrap(); - let device = renamed.device.as_mut().unwrap(); - device.device_name = "新设备名".to_string(); - device.platform = Some("ios".to_string()); - process_message( - &discovery.inner, - &serde_json::to_vec(&renamed).unwrap(), - second, - now + Duration::from_secs(30), - ); - let snapshots = host.snapshots.lock().unwrap(); - let last = snapshots.last().unwrap(); - assert_eq!(last[0].device_name, "新设备名"); - assert_eq!(last[0].platform.as_deref(), Some("ios")); - drop(snapshots); - - let mut peers = discovery.inner.peers.lock().unwrap(); - assert!(peers.expire(now + ADDRESS_TTL)); - assert_eq!( - peers.snapshot()[0].base_urls, - ["https://192.168.1.20:56000"] - ); - assert!(peers.expire(now + ADDRESS_TTL + Duration::from_secs(30))); - assert!(peers.snapshot().is_empty()); - } -} +mod tests; diff --git a/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/apple.rs b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/apple.rs new file mode 100644 index 00000000..25be3b10 --- /dev/null +++ b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/apple.rs @@ -0,0 +1,169 @@ +use std::collections::{HashMap, HashSet}; +use std::sync::Weak; +use std::time::Duration; + +use async_dnssd::{ + BrowseData, BrowseResult, BrowsedFlags, RegisterData, Registration, ResolveResult, + ResolvedHostFlags, ScopedSocketAddr, TxtRecord, +}; +use futures_util::future::{AbortHandle, Abortable}; +use futures_util::stream::FuturesUnordered; +use futures_util::{StreamExt, TryStreamExt}; +use tokio::task::JoinHandle; +use tt_contracts::lan_discovery::LanDiscoveryAnnouncement; +use tt_domain::errors::DomainError; + +use super::{ + DiscoveryInner, MAX_DEVICES, SERVICE_TYPE, decode_device, discovery_error, txt_properties, + update_peer, +}; + +pub(super) struct Backend { + registration: Option<(u16, Registration)>, +} + +pub(super) fn start( + inner: Weak, +) -> Result<(Backend, JoinHandle>), DomainError> { + Ok((Backend { registration: None }, tokio::spawn(browse(inner)))) +} + +impl Backend { + pub(super) async fn announce( + &mut self, + device: &LanDiscoveryAnnouncement, + ) -> Result<(), DomainError> { + let mut txt = TxtRecord::new(); + for (key, value) in txt_properties(device) { + txt.set_value(key.as_bytes(), &value) + .map_err(|error| discovery_error(format!("{error:?}")))?; + } + if let Some((port, registration)) = &self.registration + && *port == device.port + { + return registration + .get_default_txt_record() + .update_record(txt.rdata(), 0) + .map_err(discovery_error); + } + self.registration = None; + let registering = async_dnssd::register_extended( + SERVICE_TYPE, + device.port, + RegisterData { + name: Some(device.device_id.as_str()), + domain: Some("local."), + txt: txt.rdata(), + ..Default::default() + }, + ) + .map_err(discovery_error)?; + // A permission prompt or failed registration must not hold up the HTTPS server. + let (registration, _) = tokio::time::timeout(Duration::from_secs(3), registering) + .await + .map_err(|_| discovery_error("Timed out announcing this device"))? + .map_err(discovery_error)?; + self.registration = Some((device.port, registration)); + Ok(()) + } + + pub(super) async fn stop(&mut self) -> Result<(), DomainError> { + self.registration = None; + Ok(()) + } +} + +async fn browse(inner: Weak) -> Result<(), DomainError> { + let mut browser = async_dnssd::browse_extended( + SERVICE_TYPE, + BrowseData { + domain: Some("local."), + ..Default::default() + }, + ); + let mut handles = HashMap::::new(); + let mut services = FuturesUnordered::new(); + loop { + tokio::select! { + result = browser.try_next() => { + let service = result.map_err(discovery_error)? + .ok_or_else(|| discovery_error("Bonjour browser stopped"))?; + let source = format!("{}@{}", service.service_name, service.interface.into_raw()); + if !service.flags.contains(BrowsedFlags::ADD) { + if let Some(handle) = handles.remove(&source) { + handle.abort(); + } + update_peer(&inner, &source, None); + } else if !handles.contains_key(&source) && handles.len() < MAX_DEVICES { + let (handle, registration) = AbortHandle::new_pair(); + handles.insert(source.clone(), handle); + let inner = inner.clone(); + services.push(Abortable::new(async move { + let result = watch_service(service, &source, &inner).await; + (source, result) + }, registration)); + } + } + Some(result) = services.next(), if !services.is_empty() => { + if let Ok((source, result)) = result { + handles.remove(&source); + update_peer(&inner, &source, None); + if let Err(error) = result { + tracing::warn!("Cannot resolve Bonjour service {source}: {error}"); + } + } + } + } + } +} + +async fn watch_service( + service: BrowseResult, + source: &str, + inner: &Weak, +) -> Result<(), DomainError> { + let mut resolving = service.resolve(); + let mut resolved = resolving + .try_next() + .await + .map_err(discovery_error)? + .ok_or_else(|| discovery_error("Bonjour service resolution stopped"))?; + let mut device = decode_resolved(&resolved)?; + let mut addresses = resolved.resolve_socket_address(); + let mut ips = HashSet::new(); + loop { + tokio::select! { + result = resolving.try_next() => { + let next = result.map_err(discovery_error)? + .ok_or_else(|| discovery_error("Bonjour service resolution stopped"))?; + device = decode_resolved(&next)?; + if next.host_target != resolved.host_target || next.port != resolved.port { + ips.clear(); + addresses = next.resolve_socket_address(); + } + resolved = next; + } + result = addresses.try_next() => { + let address = result.map_err(discovery_error)? + .ok_or_else(|| discovery_error("Bonjour address resolution stopped"))?; + let ScopedSocketAddr::V4 { address: ip, .. } = address.address else { continue }; + if address.flags.contains(ResolvedHostFlags::ADD) { + ips.insert(ip); + } else { + ips.remove(&ip); + } + } + } + update_peer( + inner, + source, + Some((device.clone(), ips.iter().copied().collect())), + ); + } +} + +fn decode_resolved(service: &ResolveResult) -> Result { + let txt = TxtRecord::parse(&service.txt) + .ok_or_else(|| DomainError::InvalidData("Invalid Bonjour TXT record".to_string()))?; + decode_device(service.port, |key| txt.get(key.as_bytes()).flatten()) +} diff --git a/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/mdns.rs b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/mdns.rs new file mode 100644 index 00000000..d2c97f3c --- /dev/null +++ b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/mdns.rs @@ -0,0 +1,100 @@ +use std::sync::Weak; + +use mdns_sd::{DaemonEvent, IfKind, ServiceDaemon, ServiceEvent, ServiceInfo, TxtProperty}; +use tokio::task::JoinHandle; +use tt_contracts::lan_discovery::LanDiscoveryAnnouncement; +use tt_domain::errors::DomainError; + +use super::{ + DiscoveryInner, SERVICE_TYPE, decode_device, discovery_error, txt_properties, update_peer, +}; + +pub(super) struct Backend { + daemon: ServiceDaemon, +} + +pub(super) fn start( + inner: Weak, +) -> Result<(Backend, JoinHandle>), DomainError> { + let backend = Backend { + daemon: ServiceDaemon::new().map_err(discovery_error)?, + }; + // The existing HTTPS listener and return-address selection use IPv4. + backend + .daemon + .disable_interface(vec![IfKind::IPv6, IfKind::LoopbackV4]) + .map_err(discovery_error)?; + let events = backend + .daemon + .browse(&format!("{SERVICE_TYPE}.local.")) + .map_err(discovery_error)?; + let monitor = backend.daemon.monitor().map_err(discovery_error)?; + let receiver = tokio::spawn(async move { + loop { + tokio::select! { + result = events.recv_async() => { + match result.map_err(discovery_error)? { + ServiceEvent::ServiceResolved(service) => { + let device = decode_device(service.get_port(), |key| service.get_property_val(key).flatten()); + match device { + Ok(device) => update_peer(&inner, service.get_fullname(), Some((device, service.get_addresses_v4().into_iter().collect()))), + Err(error) => { + update_peer(&inner, service.get_fullname(), None); + tracing::warn!("Ignoring invalid LAN mDNS service: {error}"); + } + } + } + ServiceEvent::ServiceRemoved(_, name) => update_peer(&inner, &name, None), + _ => {} + } + } + result = monitor.recv_async() => { + if let DaemonEvent::Error(error) = result.map_err(discovery_error)? { + // One interface failing does not make devices on other interfaces unusable. + tracing::warn!("LAN mDNS discovery: {error}"); + } + } + } + } + }); + Ok((backend, receiver)) +} + +impl Backend { + pub(super) async fn announce( + &mut self, + device: &LanDiscoveryAnnouncement, + ) -> Result<(), DomainError> { + let properties = txt_properties(device) + .iter() + .map(|(key, value)| TxtProperty::from((*key, value.as_slice()))) + .collect::>(); + let service = ServiceInfo::new( + &format!("{SERVICE_TYPE}.local."), + device.device_id.as_str(), + &format!("tt-{}.local.", device.device_id.as_str()), + (), + device.port, + properties, + ) + .map_err(discovery_error)? + .enable_addr_auto(); + self.daemon.register(service).map_err(discovery_error) + } + + pub(super) async fn stop(&mut self) -> Result<(), DomainError> { + self.daemon + .shutdown() + .map_err(discovery_error)? + .recv_async() + .await + .map_err(discovery_error)?; + Ok(()) + } +} + +impl Drop for Backend { + fn drop(&mut self) { + let _ = self.daemon.shutdown(); + } +} diff --git a/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/tests.rs b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/tests.rs new file mode 100644 index 00000000..476f445b --- /dev/null +++ b/src-tauri/crates/tt-adapter-sync/src/sync/lan/peer_discovery/tests.rs @@ -0,0 +1,164 @@ +use super::*; + +#[derive(Default)] +struct RecordingHost { + snapshots: Mutex>>, + changed: tokio::sync::Notify, +} + +#[async_trait] +impl LanDiscoveryHost for RecordingHost { + async fn set_enabled(&self, _enabled: bool) -> Result<(), DomainError> { + Ok(()) + } + + fn publish_changed(&self, devices: Vec) { + self.snapshots.lock().unwrap().push(devices); + self.changed.notify_one(); + } +} + +fn announcement(name: &str) -> LanDiscoveryAnnouncement { + LanDiscoveryAnnouncement { + device_id: DeviceId::new(uuid::Uuid::new_v4().to_string()).unwrap(), + device_name: name.to_string(), + platform: Some("android".to_string()), + port: 56000, + spki_sha256: "a".repeat(43), + } +} + +#[test] +fn discovery_preserves_unicode_names_and_remaining_interfaces_when_a_service_disappears() { + let host = Arc::new(RecordingHost::default()); + let discovery = LanPeerDiscovery::new(host.clone()); + let inner = Arc::downgrade(&discovery.inner); + let original = announcement(&"🦀".repeat(64)); + let mut txt = txt_properties(&original) + .into_iter() + .collect::>(); + let device = decode_device(original.port, |key| txt.get(key).map(Vec::as_slice)).unwrap(); + assert_eq!(device.device_name, original.device_name); + txt.remove("spki"); + assert!(decode_device(original.port, |key| txt.get(key).map(Vec::as_slice)).is_err()); + + let first = vec!["192.168.1.10".parse().unwrap()]; + let second = vec!["192.168.1.20".parse().unwrap()]; + update_peer(&inner, "wifi", Some((device.clone(), first.clone()))); + update_peer(&inner, "wifi", Some((device.clone(), first))); + assert_eq!(host.snapshots.lock().unwrap().len(), 1); + update_peer(&inner, "ethernet", Some((device.clone(), second.clone()))); + assert_eq!( + discovery.candidate_urls(&device.device_id), + ["https://192.168.1.10:56000", "https://192.168.1.20:56000",] + ); + + let mut renamed = device.clone(); + renamed.device_name = "新设备名".to_string(); + renamed.platform = Some("ios".to_string()); + update_peer(&inner, "ethernet", Some((renamed, second))); + let snapshot = discovery.snapshot(); + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot[0].device_name, "新设备名"); + assert_eq!(snapshot[0].platform.as_deref(), Some("ios")); + update_peer(&inner, "wifi", None); + assert_eq!( + discovery.candidate_urls(&device.device_id), + ["https://192.168.1.20:56000"] + ); + update_peer(&inner, "ethernet", None); + assert!(discovery.snapshot().is_empty()); +} + +#[cfg(target_os = "macos")] +#[path = "mdns.rs"] +mod mdns; + +#[cfg(target_os = "macos")] +async fn wait_for_device( + discovery: &LanPeerDiscovery, + host: &RecordingHost, + id: &DeviceId, + name: Option<&str>, +) { + tokio::time::timeout(Duration::from_secs(8), async { + loop { + let changed = host.changed.notified(); + let devices = discovery.snapshot(); + let device = devices.iter().find(|device| &device.device_id == id); + if match name { + Some(name) => device.is_some_and(|device| device.device_name == name), + None => device.is_none(), + } { + break; + } + changed.await; + } + }) + .await + .expect("device discovery update"); +} + +#[cfg(target_os = "macos")] +#[tokio::test] +#[ignore = "requires access to local multicast networking"] +async fn system_bonjour_and_portable_mdns_discover_rename_and_remove_each_other() { + let apple_host = Arc::new(RecordingHost::default()); + let portable_host = Arc::new(RecordingHost::default()); + let apple = LanPeerDiscovery::new(apple_host.clone()); + let portable = LanPeerDiscovery::new(portable_host.clone()); + let apple_device = announcement("Apple discovery test"); + let mut portable_device = announcement("Portable discovery test"); + let (mut backend, receiver) = mdns::start(Arc::downgrade(&portable.inner)).unwrap(); + + apple.set_local_device(apple_device.clone()).await.unwrap(); + backend.announce(&portable_device).await.unwrap(); + wait_for_device( + &apple, + &apple_host, + &portable_device.device_id, + Some(&portable_device.device_name), + ) + .await; + wait_for_device( + &portable, + &portable_host, + &apple_device.device_id, + Some(&apple_device.device_name), + ) + .await; + + portable_device.device_name = "🦀".repeat(64); + backend.announce(&portable_device).await.unwrap(); + wait_for_device( + &apple, + &apple_host, + &portable_device.device_id, + Some(&portable_device.device_name), + ) + .await; + apple.set_device_name("已改名的 Apple 设备").await.unwrap(); + wait_for_device( + &portable, + &portable_host, + &apple_device.device_id, + Some("已改名的 Apple 设备"), + ) + .await; + apple.stop().await.unwrap(); + wait_for_device(&portable, &portable_host, &apple_device.device_id, None).await; + + apple.discover_devices().await.unwrap(); + wait_for_device( + &apple, + &apple_host, + &portable_device.device_id, + Some(&portable_device.device_name), + ) + .await; + receiver.abort(); + let _ = receiver.await; + backend.stop().await.unwrap(); + wait_for_device(&apple, &apple_host, &portable_device.device_id, None).await; + apple.stop().await.unwrap(); +} diff --git a/src-tauri/crates/tt-application/src/services/lan_sync_service.rs b/src-tauri/crates/tt-application/src/services/lan_sync_service.rs index ae988664..1d1c7a45 100644 --- a/src-tauri/crates/tt-application/src/services/lan_sync_service.rs +++ b/src-tauri/crates/tt-application/src/services/lan_sync_service.rs @@ -225,6 +225,10 @@ impl LanSyncService { self.discovery.discover_devices().await } + pub async fn refresh_discovery_if_started(&self) -> Result<(), DomainError> { + self.discovery.refresh_if_started().await + } + pub async fn set_device_name(&self, name: &str) -> Result<(), DomainError> { let name = name.trim(); validate_device_name(name)?; diff --git a/src-tauri/crates/tt-application/src/services/lan_sync_service/tests.rs b/src-tauri/crates/tt-application/src/services/lan_sync_service/tests.rs index b48fecb0..22209623 100644 --- a/src-tauri/crates/tt-application/src/services/lan_sync_service/tests.rs +++ b/src-tauri/crates/tt-application/src/services/lan_sync_service/tests.rs @@ -265,6 +265,10 @@ struct NoopDeviceDiscovery; #[async_trait] impl LanDeviceDiscovery for NoopDeviceDiscovery { + async fn refresh_if_started(&self) -> Result<(), DomainError> { + Ok(()) + } + async fn set_device_name(&self, _name: &str) -> Result<(), DomainError> { Ok(()) } @@ -419,6 +423,10 @@ impl LanPairingClient for ManualPairingClient { #[async_trait] impl LanDeviceDiscovery for ManualPairingClient { + async fn refresh_if_started(&self) -> Result<(), DomainError> { + Ok(()) + } + async fn set_device_name(&self, _name: &str) -> Result<(), DomainError> { Ok(()) } diff --git a/src-tauri/crates/tt-ports/src/lan_discovery.rs b/src-tauri/crates/tt-ports/src/lan_discovery.rs index b65aa7d1..34b5cdfd 100644 --- a/src-tauri/crates/tt-ports/src/lan_discovery.rs +++ b/src-tauri/crates/tt-ports/src/lan_discovery.rs @@ -6,6 +6,7 @@ use tt_domain::errors::DomainError; pub trait LanDeviceDiscovery: Send + Sync { async fn discover_devices(&self) -> Result, DomainError>; async fn set_device_name(&self, name: &str) -> Result<(), DomainError>; + async fn refresh_if_started(&self) -> Result<(), DomainError>; } #[async_trait]