diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 0df867a..57b5f1b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -52,7 +52,8 @@ main thread │ ├── URB receive thread (reads USB/IP commands from client) │ ├── URB submit thread (forwards to physical USB device) │ └── URB reply thread (sends responses back to client) - └── device monitor thread (libusb hotplug callbacks) + └── device monitor (deferred to v1.1 per ADR-0005: libusb hotplug callbacks; + poll-based HotplugMonitor was removed — issue #29) ``` ### Client Threads @@ -271,15 +272,21 @@ battery_optimization_bypass = true ## Dependency Map ``` -usbip-core (Rust, no_std capable) +usbip-core (Rust) ├── byteorder -├── crc32fast ├── zerocopy (safe transmutes) -└── thiserror +├── crc32fast +├── thiserror +├── tracing (structured logging) +├── ring (AES-GCM, X25519, HKDF) +├── uuid (correlation IDs) +├── crossbeam +├── rand +└── tokio (async message framing) usbip-server (Rust) ├── usbip-core -├── libusb (via rusb) +├── rusb (libusb) ├── mdns-sd ├── tokio (async runtime) ├── tracing (structured logging) diff --git a/CLAUDE.md b/CLAUDE.md index 46d398a..c2c792a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -18,11 +18,11 @@ cargo fmt --all -- --check # rustfmt is configured — s cargo clippy --workspace -- -D warnings # CI gate; treat warnings as errors ``` -Per-crate: +Per-crate (all three crates have integration tests under `tests/` plus inline `#[cfg(test)]` modules; see `find . -name '*.rs' -path '*/tests/*'` and `grep -rl '#[cfg(test)]'`): ```bash -cargo test --release -p usbip-core # only crate with real coverage today -cargo test --release -p usbip-server # passes vacuously -cargo test --release -p usbip-client # passes vacuously +cargo test --release -p usbip-core # protocol types + descriptor fixtures +cargo test --release -p usbip-server # wire protocol, REST API, encryption +cargo test --release -p usbip-client # VHCI injection seam + inline URB forwarding ``` Android (committed wrapper exists — `./gradlew`, not bare `gradle`): diff --git a/CONTEXT.md b/CONTEXT.md index 0fc450f..aa41d46 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -86,3 +86,41 @@ _Avoid_: Admin port, management port, HTTP port UDP port for `_usbip._tcp.local` service advertisements. Link-local only — does not cross subnets or VLANs. _Avoid_: Discovery port, broadcast port + +## Module ownership + +This section names ONE owner per architectural concern. These claims are +authoritative; every other doc (CLAUDE.md, ARCHITECTURE.md, PROTOCOL.md, +ROADMAP.md, README.md, docs/*.md) must agree with them. When a doc disagrees, +the doc is wrong — fix the doc, not this section. + +- **ConfigOwner**: `Server::app_state` (server/usbip-server/src/server.rs) — + the one place that reads and writes the merged config (`api::load_config` / + `AppState.config`). CLI flags in `main.rs` seed field values; + `app_state` reconciles them with the persisted file. +- **MetricsOwner**: `server/usbip-server/src/metrics.rs` — + `build_metrics_router` and the `ENCRYPTION_ENABLED` gauge are live (wired + from `main.rs`) and are the ONLY metrics present. The former + `DEVICES_EXPORTED` / `CLIENTS_CONNECTED` / `URB_SUBMIT_TOTAL` / + `URB_BYTES_TOTAL` statics were DELETED (issue #31): they were never wired. + If live export/connected/URB counters are ever needed again, add them at + the single site that owns each event, not as dead statics. +- **VhciBackendOwner**: `Client` (client/usbip-client/src/client.rs) — the + one place that constructs a `VhciBackend` (`detect_backend`), + and the injection seam `Client::new_with_vhci`. +- **HotplugOwner**: DEFERRED (issue #29). The poll-based `HotplugMonitor` + was DELETED (Path B); no v1.0 hotplug detection exists. ADR-0005 remains + the v1.1 blueprint (libusb/callback-based). Any future monitor is owned by + `Server` and must follow ADR-0005. +- **DiscoveryTxtOwner**: `usbip-core::discovery_txt` — + `shared/usbip-core/src/discovery_txt.rs` owns the `device` TXT wire format + (encode + decode: vid/pid/bus/name tuples, e.g. + `vid=0x046d,pid=0xc261,bus=1-1,n=046d:c261`). The server + (`discovery.rs`) advertises and delegates encode to it; the client + (`usbip-client::discovery`) decodes via it (issue #30). One seam. +- **AndroidStateOwner**: single Kotlin class managing Android connection / + runtime state — pending issue #32 (nominating the class is part of that work). +- **AndroidComposerOwner**: single Kotlin class composing the Compose UI + tree (phone and TV) — pending issue #32. +- **AndroidControllerOwner**: single Kotlin class owning UI event dispatch + and state mutation — pending issue #32. diff --git a/PROTOCOL.md b/PROTOCOL.md index 8f87b83..c0e645c 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -245,24 +245,29 @@ When encryption is enabled, after the initial TCP handshake: ``` Client Server │ │ - │ ECDH key exchange (X25519) │ + │ [4-byte len=32][X25519 pubkey] │ │────────────────────────────────────────►│ │ │ - │ ECDH response │ + │ [4-byte len=32][X25519 pubkey] │ │◄────────────────────────────────────────│ │ │ │ HKDF-SHA256 → AES-256-GCM session key │ │ │ │ All subsequent USB/IP messages: │ - │ [4-byte ciphertext length][ciphertext] │ + │ [4-byte ciphertext length BE] │ + │ [ciphertext || 12-byte nonce || │ + │ 16-byte GCM tag] │ │ Each message has unique 96-bit nonce │ - │ (initialized from session key + seq) │ │ │ ``` -## 7. Compression Extension (optional) +The 4-byte ciphertext length prefix is big-endian. The nonce (12 bytes) and +GCM tag (16 bytes) are appended to the ciphertext within that length. -For bulk endpoints on slow links: +## 7. Compression Extension (optional — NOT implemented) + +Designed for bulk endpoints on slow links. Not yet implemented in code; +this section describes the intended wire format for a future release. ``` After encryption (if enabled): diff --git a/README.md b/README.md index bf93121..636caf8 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,7 @@ Open the app on both — auto-discovery via mDNS. Tap the device on TV to connec - **USB/IP protocol** — same as Linux kernel, battle-tested since 2008. - **mDNS discovery** — no IP config needed, devices find each other. - **AES-256-GCM encryption** — optional, for untrusted networks. -- **Sub-1ms per-URB latency** on Ethernet, 2-5ms on Wi-Fi 6. +- **Sub-2ms per-URB latency target** on Ethernet, 2-5ms on Wi-Fi 6. - **Service mode** — runs headless, survives reboots. - **Android TV UI** — D-pad navigable, remote-friendly. - **Auto-reconnect** — survives network flaps and device cycles. @@ -75,9 +75,9 @@ See [ARCHITECTURE.md](ARCHITECTURE.md). ## Latency Budget -HID URB round-trip on gigabit Ethernet: **~700 µs total RTT**. +HID URB round-trip target on gigabit Ethernet: **~1.5-3.5 ms total RTT** (see [docs/PERFORMANCE.md](docs/PERFORMANCE.md) for the breakdown). -For FFB at 250 Hz (<4ms needed): 5x headroom on Ethernet, 2x on good Wi-Fi. +For FFB at 250 Hz (<4ms needed): comfortable headroom on Ethernet, marginal on Wi-Fi. --- diff --git a/ROADMAP.md b/ROADMAP.md index f0eabdd..8a5c308 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -109,7 +109,7 @@ Project roadmap and milestone tracking for the USB/IP passthrough system. - [x] ROADMAP.md — milestone tracking - [x] docs/SETUP.md — platform setup guides - [x] docs/TROUBLESHOOTING.md — diagnosis and fixes -- [x] docs/DEVICES.md — device profiles and known quirks +- [ ] docs/DEVICES.md — device profiles and known quirks (file not yet created) - [x] docs/ANDROID-TV.md — TV-specific guide - [x] docs/PERFORMANCE.md — latency, benchmarks, tuning - [x] docs/BUILDING.md — compilation from source @@ -163,11 +163,11 @@ not just the original reference hardware. Delivered per PRD #1. ## Milestone 13: Reliability ✅ - [x] Structured errors with correlation IDs (per ADR-0003: must come first) -- [x] Hot-plug detection (device attach/detach after server start) +- [ ] Hot-plug detection (device attach/detach after server start) — v1.1; poll-based `hotplug.rs` was removed (issue #29), design per ADR-0005 - [x] Auto-reconnect (survive network flaps and server restarts) - [x] Multiple simultaneous client connections to different devices - [x] Linux client daemon (systemd unit + local control socket) -- [x] End-to-end latency monitoring dashboard +- [x] End-to-end latency monitoring dashboard — `/api/events` WebSocket broadcast + per-URB high-water-mark latency sampling (issue #33) ## Milestone 14: Ecosystem ✅ @@ -185,7 +185,6 @@ Features explicitly deferred beyond v1.0: - [ ] Session persistence (resume active import after server crash) — per ADR-0003 - [ ] USB 3.0 SuperSpeed support (up to 5 Gbps) - [ ] IPv6 support -- [ ] Prometheus metrics endpoint - [ ] Bandwidth throttling per client - [ ] Custom embedded firmware image (Buildroot/Yocto) - [ ] Home Assistant add-on diff --git a/client/usbip-client/src/client.rs b/client/usbip-client/src/client.rs index 9e78db9..fcdd1b9 100644 --- a/client/usbip-client/src/client.rs +++ b/client/usbip-client/src/client.rs @@ -482,11 +482,8 @@ async fn urb_forwarding_loop(stream: &mut TcpStream, vhci: &dyn VhciBackend) -> UsbIpError::from(ErrorKind::Protocol("invalid RET_SUBMIT".into())) })?; - let in_data = if ret.has_data() { - &payload[UsbIpRetSubmit::HEADER_SIZE..] - } else { - &[][..] - }; + let in_data = + if ret.has_data() { &payload[UsbIpRetSubmit::HEADER_SIZE..] } else { &[][..] }; // Complete the URB on the VHCI side vhci.complete_urb( @@ -501,10 +498,10 @@ async fn urb_forwarding_loop(stream: &mut TcpStream, vhci: &dyn VhciBackend) -> if payload.len() < UsbIpRetUnlink::SIZE { return Err(UsbIpError::from(ErrorKind::Protocol("invalid RET_UNLINK".into()))); } - let (unlink, _) = UsbIpRetUnlink::read_from_prefix(&payload[..UsbIpRetUnlink::SIZE]) - .map_err(|_| { - UsbIpError::from(ErrorKind::Protocol("invalid RET_UNLINK".into())) - })?; + let (unlink, _) = UsbIpRetUnlink::read_from_prefix( + &payload[..UsbIpRetUnlink::SIZE], + ) + .map_err(|_| UsbIpError::from(ErrorKind::Protocol("invalid RET_UNLINK".into())))?; vhci.cancel_urb(unlink.seqnum(), unlink.devid())?; }, diff --git a/client/usbip-client/src/discovery.rs b/client/usbip-client/src/discovery.rs index 7c7bf55..6a3de2f 100644 --- a/client/usbip-client/src/discovery.rs +++ b/client/usbip-client/src/discovery.rs @@ -1,12 +1,16 @@ //! mDNS service discovery for USB/IP. //! //! Browses `_usbip._tcp.local` to find servers on the local network. +//! Also provides helpers to decode the `devices` TXT key carried in +//! advertisements, delegating to `usbip_core::discovery_txt`. use mdns_sd::{ServiceDaemon, ServiceEvent}; use std::net::SocketAddr; use std::time::Duration; use tracing::info; +use usbip_core::discovery_txt::decode_devices_txt; +pub use usbip_core::discovery_txt::DiscoveredDevice; use usbip_core::error::*; pub struct MdnsBrowser { @@ -62,3 +66,75 @@ impl MdnsBrowser { Ok(servers) } } + +/// Decode the `devices` TXT value from a discovered server into a list of +/// [`DiscoveredDevice`] entries. +/// +/// This is a thin wrapper around [`usbip_core::discovery_txt::decode_devices_txt`]. +pub fn decode_devices_from_txt(txt: &str) -> UsbIpResult> { + decode_devices_txt(txt) +} + +/// Decode the `devices` TXT value from a server's TXT properties map. +/// +/// Looks up the `"devices"` key; returns an empty list if the key is +/// absent or empty. +pub fn decode_devices_from_properties( + properties: &std::collections::HashMap, +) -> UsbIpResult> { + match properties.get("devices") { + Some(txt) if !txt.is_empty() => decode_devices_txt(txt), + _ => Ok(Vec::new()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decode_devices_from_txt_single() { + let result = decode_devices_from_txt("vid=0x1234,pid=0x5678,bus=1-1,n=1234:5678").unwrap(); + assert_eq!(result.len(), 1); + assert_eq!(result[0].vid, 0x1234); + assert_eq!(result[0].pid, 0x5678); + assert_eq!(result[0].bus, "1-1"); + assert_eq!(result[0].name, "1234:5678"); + } + + #[test] + fn decode_devices_from_txt_empty() { + let result = decode_devices_from_txt("").unwrap(); + assert!(result.is_empty()); + } + + #[test] + fn decode_devices_from_txt_malformed() { + assert!(decode_devices_from_txt("vid=0x1234").is_err()); + } + + #[test] + fn decode_devices_from_properties_with_devices() { + let mut props = std::collections::HashMap::new(); + props + .insert("devices".to_string(), "vid=0x046d,pid=0xc261,bus=1-1,n=046d:c261".to_string()); + let result = decode_devices_from_properties(&props).unwrap(); + assert_eq!(result.len(), 1); + assert_eq!(result[0].vid, 0x046d); + } + + #[test] + fn decode_devices_from_properties_no_devices_key() { + let props = std::collections::HashMap::new(); + let result = decode_devices_from_properties(&props).unwrap(); + assert!(result.is_empty()); + } + + #[test] + fn decode_devices_from_properties_empty_devices() { + let mut props = std::collections::HashMap::new(); + props.insert("devices".to_string(), String::new()); + let result = decode_devices_from_properties(&props).unwrap(); + assert!(result.is_empty()); + } +} diff --git a/client/usbip-client/src/lib.rs b/client/usbip-client/src/lib.rs index e211faf..f2e2f45 100644 --- a/client/usbip-client/src/lib.rs +++ b/client/usbip-client/src/lib.rs @@ -7,5 +7,6 @@ pub mod reconnect; pub mod vhci; pub use client::{Client, ClientConfig}; +pub use discovery::{decode_devices_from_properties, decode_devices_from_txt, DiscoveredDevice}; pub use reconnect::{decide_reconnect, ReconnectConfig, ReconnectDecision, ReconnectState}; pub use vhci::{VhciBackend, VhciDevice}; diff --git a/docs/BUILDING.md b/docs/BUILDING.md index 6c01a47..7d44885 100644 --- a/docs/BUILDING.md +++ b/docs/BUILDING.md @@ -171,25 +171,35 @@ anyplug/ ### Crate Dependency Tree ``` -usbip-core (shared protocol types — no platform deps) -├── ring (AES-256-GCM, X25519) -├── zerocopy (safe transmutation for wire types) +usbip-core (shared protocol types — no OS-specific deps) ├── byteorder (big-endian types) +├── zerocopy (safe transmutation for wire types) ├── crc32fast (CRC-32 for descriptor verification) -└── serde/serde_json (config serialization) +├── thiserror (error types) +├── ring (AES-256-GCM, X25519, HKDF) +├── uuid (correlation IDs) +├── crossbeam +├── rand +├── tokio (async message framing) +└── tracing (structured logging) usbip-server (server binary) ├── usbip-core (protocol types) ├── rusb (libusb bindings) -├── tokio (async TCP) +├── tokio (async runtime) ├── mdns-sd (mDNS advertisement) -└── clap (CLI parsing) +├── clap (CLI parsing) +├── ring (AES-256-GCM) +├── axum/tower (REST API + WebSocket) +└── prometheus (metrics endpoint) usbip-client (client binary) ├── usbip-core (protocol types) ├── tokio (async TCP) ├── mdns-sd (mDNS browsing) -└── clap (CLI parsing) +├── clap (CLI parsing) +├── ring (AES-256-GCM) +└── winapi (Windows: SetupAPI, IOCTL) windows (Windows GUI + Service) ├── usbip-core (protocol types) diff --git a/docs/PERFORMANCE.md b/docs/PERFORMANCE.md index 75b51ae..348b5ba 100644 --- a/docs/PERFORMANCE.md +++ b/docs/PERFORMANCE.md @@ -31,7 +31,10 @@ Typical HID polling interval is **1 ms** (1000 Hz). Acceptable round-trip thresh ## End-to-End Latency Numbers -Measured with USB HID device on Linux server (i7-8700K) and Windows client (i5-12400), same gigabit Ethernet switch. +Target values for the round-trip budget, based on external packet-timing and +component estimates. The URB loop does not yet instrument per-URB latency +in process (issue #33); these figures are planning targets for the system, +not measurements produced by the project's own tooling. ### Base Latency (No Encryption) @@ -61,21 +64,26 @@ Encryption overhead is < 1 ms on CPUs with AES-NI (most x86 from 2012+). ARM CPU ## URB Pool Sizing -The URB pool (`shared/usbip-core/src/urb.rs`) pre-allocates buffers to avoid hot-path allocations. +The URB pool (`shared/usbip-core/src/pool.rs`) pre-allocates buffers to avoid hot-path allocations. The pool self-tunes its size via an EWMA of the observed URB rate. ### Default Configuration | Parameter | Default | Description | |-----------|---------|-------------| -| Pool size | 1024 | Number of pre-allocated URB buffers | -| Data capacity | 1024 | Max URB payload bytes | -| Buffer total | 1080 bytes | 56 header + 1024 data | +| Minimum pool size | 1024 | Number of pre-allocated URB buffers before growth | +| Bulk data capacity | 16 KiB | Max bulk/control URB payload bytes | +| Interrupt data capacity | 64 B | Max interrupt-transfer URB payload bytes | -**Typical HID controller:** IN URBs ~78 bytes, OUT URBs ~4 bytes, up to 1000 URBs/sec each way. Pool of 1024 = ~1 MB, cycles every ~1 sec at max rate. Latency spikes beyond capacity cause allocation on hot path. +**Typical HID controller:** IN URBs ~78 bytes, OUT URBs ~4 bytes, up to 1000 URBs/sec each way. A bulk pool of 1024 × 16 KiB buffers holds ~16 MiB and cycles every ~1 sec at max rate. Latency spikes beyond capacity cause allocation on hot path. + +> Note: `UrbBufferPool` is currently exercised only in `usbip-core`'s own +> tests/benchmarks; the server URB loop (`handle_urb_loop`) does not yet +> acquire buffers from it. These figures describe the pool's design defaults, +> not the live data path. ### When to Increase Pool Size -Increase `POOL_SIZE` if you see `allocating URB buffer on hot path` in trace logs, have high-latency USB devices (isochronous), or use encryption. +Increase the `min_size` argument if you see `allocating URB buffer on hot path` in trace logs, have high-latency USB devices (isochronous), or use encryption. ``` Target: pool holds 2x URBs seen in one second @@ -91,11 +99,12 @@ The batcher (`server/usbip-server/src/batcher.rs`) coalesces multiple URBs into | Flush interval | CPU savings | Added latency | Best for | |---------------|-------------|---------------|----------| | 100 µs | 5-10% | ~50 µs avg | Low-latency (HID, wheel) | +| 200 µs | 10-20% | ~100 µs avg | Default | | 500 µs | 20-30% | ~250 µs avg | Bulk (mass storage) | | 1 ms | 40-50% | ~500 µs avg | Throughput-sensitive | | None | 0% | 0 µs | Minimum latency | -Default is **100 µs** — balances CPU and latency for HID devices. +Default is **200 µs** — balances CPU and latency for HID devices. --- @@ -206,10 +215,11 @@ If Wi-Fi is the only option: ### Per-Message Wire Overhead -Without encryption: `[8-byte header] [payload]` = 8 + N bytes -With encryption: `[8-byte header] [12-byte nonce] [encrypted] [16-byte tag]` = 36 + N bytes +Without encryption: raw kernel framing — the USB/IP message (`[8-byte header] [payload]`) is written directly = 8 + N bytes. + +With encryption: `[4-byte ciphertext length BE][ciphertext][12-byte nonce][16-byte GCM tag]`, where ciphertext is the encrypted full USB/IP message (header + payload) = 4 + (8 + N) + 12 + 16 = 40 + N bytes on the wire. -For a 78-byte IN URB: 86 bytes unencrypted vs 114 encrypted (~32% increase). CPU cost ~1-2 µs/URB on AES-NI. +For a 78-byte IN URB (message size 86): 86 bytes unencrypted vs 126 encrypted (~47% increase). CPU cost ~1-2 µs/URB on AES-NI. ### CPU Usage at 1000 URBs/sec diff --git a/docs/adr/0006-what-not-to-borrow.md b/docs/adr/0006-what-not-to-borrow.md new file mode 100644 index 0000000..872117f --- /dev/null +++ b/docs/adr/0006-what-not-to-borrow.md @@ -0,0 +1,93 @@ +# What NOT to borrow from adjacent USB-over-network designs + +The architecture review across related projects (VirtualHere, Cemuhook, DSU +server, and the console-remote ecosystem) identified a set of tempting-but-wrong +patterns that this project must not adopt. They each solve a real problem well +enough to seem reusable, but they conflict with the project's core constraint: +**byte-for-byte passthrough of native USB descriptors and endpoints over +USB/IP.** + +## Rejected patterns + +### 1. UDP-only transports + +Some implementations (notably Cemuhook protocol and certain cloud-game +companions) carry control + input over UDP to minimise latency. This is +rejected because: + +- UDP has no reliability, ordering, or congestion control by default. + USB/IP URBs carry arbitrary bulk data (mass storage) and control transfers; + dropping or reordering a single packet corrupts the device session + irrecoverably. +- The project's latency budget (~1.5-3.5 ms round-trip on Ethernet) already + fits comfortably inside TCP's retransmission overhead; there is no latency + case for UDP. +- Encryption (AES-256-GCM) is significantly easier to do correctly on a + reliable, ordered byte stream than on a datagram protocol. + +### 2. REST-only topology + +Some designs expose device control exclusively through HTTP/REST endpoints, +treating the data plane as a side effect of API calls. This is rejected because: + +- REST is designed for request/response, not for a sustained bidirectional + byte stream. USB/IP is fundamentally a streaming protocol: the server and + client exchange URBs continuously for the lifetime of a session. +- A REST wrapper adds a serialisation boundary between the wire protocol and + the URB forwarding loop, which would force copying and re-framing of every + message — defeating zero-copy where it matters. +- The project already uses REST (`/api/*`) as the *control* plane (status, + scan, connect, config). It must not be promoted to the *data* plane. + +### 3. MSG_RUMBLE opcodes + +Some controller passthrough schemes define ad-hoc "rumble" or effect opcodes +layered on top of the transport to send force-feedback commands. This is +rejected because: + +- Force feedback is already expressed natively in USB HID reports + (output reports and feature reports). A custom rumble opcode is a second, + competing encoding of data the HID stack already produces — two sources of + truth for the same physical action. +- The whole point of USB/IP passthrough is that the importing OS loads the + real vendor/class driver; the driver already issues HID output reports. + Injecting MSG_RUMBLE lets a generic client bypass the driver, which is + only useful when emulating a device (explicitly out of scope). +- Any custom opcode must be versioned, documented, and kept in sync with + every client and server. Native HID needs none of that machinery. + +### 4. Cemuhook DSU axis layout + +Cemuhook's DSU ("DS4Windows-compatible") protocol defines its own 6-axis +layout (accel + gyro with a specific byte order and scale) that game clients +parse directly. The project must not build its own axis-layout protocol on +top of USB/IP. This is rejected because: + +- DSU is a *replacement* for the native controller protocol — the DSU server + converts the hardware's raw reports into DSU's fixed schema. That is the + opposite of passthrough. +- A custom axis layout would require each consumer (game, app) to implement + the schema, exactly duplicating what the OS HID/FFB stack already does for + native devices. +- The narrowest correct way to support input is to let the importing OS load + the real HID driver and speak standard HID reports. Any game that already + supports the device directly will work with zero extra client code. + +## Considered option + +- **"Meet the ecosystem halfway with a compatibility layer that speaks DSU + or UDP on the input path"** — rejected. It erodes the passthrough guarantee + (devices would report through two different paths), adds a second network + surface to secure, and grows a compatibility-maintenance surface that does + not advance the core USB/IP design. + +## Consequences + +- The project keeps a single data plane (TCP + USB/IP) and a single encoding + of device behaviour (native USB descriptors and HID reports). +- New protocol features must justify themselves against these anti-patterns. + If a proposed feature looks like any of the four rejected patterns, it + needs a stronger reason than "it's what adjacent products ship." +- The RetroPie / Lakka / Steam Link ecosystem packaging (see CONTEXT.md + "Ecosystem integration") operates at the *existing* USB/IP boundary, not + by adding a second protocol. diff --git a/server/usbip-server/Cargo.toml b/server/usbip-server/Cargo.toml index 7bba393..a8545f1 100644 --- a/server/usbip-server/Cargo.toml +++ b/server/usbip-server/Cargo.toml @@ -57,3 +57,7 @@ tokio-tungstenite = "0.24" [[bench]] name = "forward_bench" harness = false + +[[bench]] +name = "urb_high_water_mark" +harness = false diff --git a/server/usbip-server/benches/urb_high_water_mark.rs b/server/usbip-server/benches/urb_high_water_mark.rs new file mode 100644 index 0000000..b5c6007 --- /dev/null +++ b/server/usbip-server/benches/urb_high_water_mark.rs @@ -0,0 +1,178 @@ +//! Benchmarks for URB latency high-water-mark CAS reduction. +//! +//! Compares naive (every-sample CAS + broadcast) vs. HWM +//! (only-on-new-peak) approaches. Reports CAS count reduction. +//! +//! Run with: cargo bench -p usbip-server -- urb_high_water_mark + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use criterion::{black_box, criterion_group, criterion_main, Criterion}; +use tokio::sync::broadcast; + +use usbip_server::api::LatencySample; + +const URBS_PER_THREAD: usize = 10_000; +const NUM_THREADS: usize = 4; + +/// Naive approach: every sample performs a load + conditional CAS on +/// the shared atomic and broadcasts on CAS success. The counter +/// tracks how many times the shared atomic was touched. +fn naive_bench(latencies: &[u64]) -> u64 { + let global = Arc::new(AtomicU64::new(0)); + let (tx, _rx) = broadcast::channel::(1024); + let atomic_ops = Arc::new(AtomicU64::new(0)); + + std::thread::scope(|s| { + for thread_id in 0..NUM_THREADS { + let g = Arc::clone(&global); + let t = tx.clone(); + let ops = Arc::clone(&atomic_ops); + s.spawn(move || { + let base = thread_id as u64 * 1000; + for (i, &lat) in latencies.iter().enumerate() { + let sample_us = base + lat; + // Naive: every URB touches the shared atomic. + let current = g.load(Ordering::Relaxed); + ops.fetch_add(1, Ordering::Relaxed); + if sample_us > current { + if g.compare_exchange_weak( + current, + sample_us, + Ordering::AcqRel, + Ordering::Relaxed, + ) + .is_ok() + { + let _ = t.send(LatencySample { + latency_us: sample_us, + device: format!("bench-{thread_id}"), + seqnum: (i * NUM_THREADS + thread_id) as u64, + }); + } + } + } + }); + } + }); + + Arc::try_unwrap(atomic_ops).unwrap().into_inner() +} + +/// HWM approach: thread-local peak filters out most URBs. Only +/// attempts CAS when local peak exceeds the global peak. +fn hwm_bench(latencies: &[u64]) -> u64 { + let global = Arc::new(AtomicU64::new(0)); + let (tx, _rx) = broadcast::channel::(1024); + let atomic_ops = Arc::new(AtomicU64::new(0)); + + std::thread::scope(|s| { + for thread_id in 0..NUM_THREADS { + let g = Arc::clone(&global); + let t = tx.clone(); + let ops = Arc::clone(&atomic_ops); + s.spawn(move || { + let base = thread_id as u64 * 1000; + let mut local_peak: u64 = 0; + for (i, &lat) in latencies.iter().enumerate() { + let sample_us = base + lat; + // HWM: thread-local check -- no atomic access. + if sample_us <= local_peak { + continue; + } + local_peak = sample_us; + + // Only reach the shared atomic when local peak + // is a new contender. + let current_global = g.load(Ordering::Relaxed); + ops.fetch_add(1, Ordering::Relaxed); + if sample_us <= current_global { + continue; + } + + if g.compare_exchange_weak( + current_global, + sample_us, + Ordering::AcqRel, + Ordering::Relaxed, + ) + .is_ok() + { + let _ = t.send(LatencySample { + latency_us: sample_us, + device: format!("bench-{thread_id}"), + seqnum: (i * NUM_THREADS + thread_id) as u64, + }); + } else { + local_peak = current_global; + } + } + }); + } + }); + + Arc::try_unwrap(atomic_ops).unwrap().into_inner() +} + +/// Generate synthetic URB latencies that oscillate around a baseline, +/// simulating steady-state USB traffic with occasional spikes. +fn synthetic_latencies() -> Vec { + let mut lats = Vec::with_capacity(URBS_PER_THREAD); + for i in 0..URBS_PER_THREAD { + let jitter = ((i * 7 + 13) % 30) as u64; + let spike = if i % 500 == 0 { 500 } else { 0 }; + lats.push(100 + jitter + spike); + } + lats +} + +fn bench_hwm_cas_reduction(c: &mut Criterion) { + let latencies = synthetic_latencies(); + let total_urbs = (URBS_PER_THREAD * NUM_THREADS) as u64; + + let mut group = c.benchmark_group("latency_hwm"); + group + .measurement_time(Duration::from_secs(3)) + .warm_up_time(Duration::from_secs(1)) + .sample_size(10); + + group.bench_function("naive_cas_attempts", |b| { + b.iter(|| { + let count = naive_bench(black_box(&latencies)); + black_box(count); + }) + }); + + group.bench_function("hwm_cas_attempts", |b| { + b.iter(|| { + let count = hwm_bench(black_box(&latencies)); + black_box(count); + }) + }); + + group.finish(); + + // Post-bench assertion: HWM touches the shared atomic far less. + let naive = naive_bench(&latencies); + let hwm = hwm_bench(&latencies); + let reduction = 100.0 * (1.0 - (hwm as f64 / naive as f64)); + eprintln!("\nCAS reduction: naive={naive} hwm={hwm} ({reduction:.1}% reduction)"); + assert!(hwm < naive, "HWM ({hwm}) should have fewer CAS ops than naive ({naive})"); + assert!( + hwm <= total_urbs / 2, + "HWM ({hwm}) should be well below half of total URBs ({total_urbs})" + ); +} + +criterion_group! { + name = hwm_benches; + config = Criterion::default() + .measurement_time(Duration::from_secs(3)) + .warm_up_time(Duration::from_secs(1)) + .sample_size(10); + targets = bench_hwm_cas_reduction, +} + +criterion_main!(hwm_benches); diff --git a/server/usbip-server/src/discovery.rs b/server/usbip-server/src/discovery.rs index ca5a19f..f08c7ca 100644 --- a/server/usbip-server/src/discovery.rs +++ b/server/usbip-server/src/discovery.rs @@ -95,26 +95,10 @@ impl MdnsAdvertiser { /// Encode the device list as the `devices` TXT value. /// -/// Format: comma-separated `vid=0xVVVV,pid=0xPPPP,bus=B-B,n=NAME` tuples. -/// The `n=` field is the human-readable device name; today -/// `UsbIpDeviceEntry` carries no product string, so we fall back to a -/// stable `VVVV:PPPP` placeholder. A future change can plumb `iProduct` -/// from the descriptor tree. +/// Delegates to `usbip_core::discovery_txt::encode_devices_txt` — the +/// single canonical seam for this wire format. pub fn encode_devices_txt(devices: &[UsbIpDeviceEntry]) -> String { - devices - .iter() - .map(|d| { - format!( - "vid=0x{:04x},pid=0x{:04x},bus={},n={:04x}:{:04x}", - d.vid(), - d.pid(), - d.busid_str(), - d.vid(), - d.pid() - ) - }) - .collect::>() - .join(",") + usbip_core::discovery_txt::encode_devices_txt(devices) } impl Drop for MdnsAdvertiser { diff --git a/server/usbip-server/src/hotplug.rs b/server/usbip-server/src/hotplug.rs deleted file mode 100644 index 11d5840..0000000 --- a/server/usbip-server/src/hotplug.rs +++ /dev/null @@ -1,262 +0,0 @@ -//! Hot-plug detection for USB device attach/detach events. -//! -//! Provides a platform-agnostic interface for monitoring USB device -//! arrival and removal. On Linux this uses libusb hotplug callbacks; -//! on other platforms it falls back to polling or is stubbed. -//! -//! Events carry structured error information with correlation IDs -//! for tracing through the system (see ADR-0003). - -use std::sync::mpsc; - -use usbip_core::error::*; - -/// Events emitted by the hotplug monitor. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum HotplugEvent { - /// A USB device was attached. - Attached { - /// The bus ID string (e.g., "3-2"). - busid: String, - /// Vendor ID. - vid: u16, - /// Product ID. - pid: u16, - }, - /// A USB device was detached. - Detached { - /// The bus ID string that was removed. - busid: String, - /// Correlation ID linking to any in-flight URB error. - correlation_id: CorrelationId, - }, -} - -/// Trait for hotplug monitors. -/// -/// Platform-specific implementations (libusb, IOKit, Windows -/// RegisterDeviceNotification, Android UsbManager) provide -/// the hook for polling or receiving callbacks. -pub trait HotplugSource: Send + 'static { - /// Poll for any pending hotplug events. - /// - /// Returns `None` when the source has been shut down. - fn poll(&mut self) -> Option; -} - -/// A no-op hotplug source that never produces events. -/// -/// Used on platforms where libusb hotplug is unavailable. -pub struct NoopHotplugSource; - -impl HotplugSource for NoopHotplugSource { - fn poll(&mut self) -> Option { - None - } -} - -/// Drives a hotplug source, emitting events into an mpsc channel. -/// -/// The monitor runs in a background thread that periodically polls -/// the source for events and forwards them to the receiver. -pub struct HotplugMonitor { - receiver: mpsc::Receiver, -} - -impl HotplugMonitor { - /// Create a new hotplug monitor from a source. - /// - /// Spawns a background thread that polls the source and forwards - /// events to the returned monitor. The thread exits when the source - /// returns `None` or when `stop` is called via the handle. - pub fn new(mut source: impl HotplugSource + 'static) -> Self { - let (tx, rx) = mpsc::channel(); - std::thread::spawn(move || { - loop { - match source.poll() { - Some(event) => { - if tx.send(event).is_err() { - // Receiver dropped, shut down. - break; - } - }, - None => { - // No event — sleep a bit before polling again. - std::thread::sleep(std::time::Duration::from_millis(100)); - }, - } - } - }); - Self { receiver: rx } - } - - /// Try to receive a hotplug event without blocking. - /// - /// Returns `None` if no event is available. - pub fn try_recv(&self) -> Option { - self.receiver.try_recv().ok() - } - - /// Block until a hotplug event is received. - /// - /// Returns `None` if the sender has been dropped (monitor shut down). - pub fn recv(&self) -> Option { - self.receiver.recv().ok() - } -} - -// ─── Tests ───────────────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - - /// A fake hotplug source that yields a fixed sequence of events. - struct FakeHotplugSource { - events: Vec>, - index: usize, - } - - impl FakeHotplugSource { - fn new(events: Vec) -> Self { - Self { events: events.into_iter().map(Some).collect(), index: 0 } - } - } - - impl HotplugSource for FakeHotplugSource { - fn poll(&mut self) -> Option { - if self.index < self.events.len() { - let event = self.events[self.index].take(); - self.index += 1; - event - } else { - // After all events are exhausted, keep returning None - // so the monitor thread stays alive. - std::thread::sleep(std::time::Duration::from_millis(10)); - None - } - } - } - - // ── HotplugEvent construction ──────────────────────────────── - - #[test] - fn test_attach_event_has_busid_and_vid_pid() { - let event = HotplugEvent::Attached { busid: "3-2".into(), vid: 0x046d, pid: 0xc261 }; - - match event { - HotplugEvent::Attached { ref busid, vid, pid } => { - assert_eq!(busid, "3-2"); - assert_eq!(vid, 0x046d); - assert_eq!(pid, 0xc261); - }, - _ => panic!("expected Attached event"), - } - } - - #[test] - fn test_detach_event_has_busid_and_correlation_id() { - let cid = CorrelationId::now_v7(); - let event = HotplugEvent::Detached { busid: "3-2".into(), correlation_id: cid }; - - match event { - HotplugEvent::Detached { ref busid, correlation_id } => { - assert_eq!(busid, "3-2"); - assert_eq!(correlation_id, cid); - }, - _ => panic!("expected Detached event"), - } - } - - // ── FakeHotplugSource ──────────────────────────────────────── - - #[test] - fn test_fake_source_returns_events_in_order() { - let events = vec![ - HotplugEvent::Attached { busid: "1-1".into(), vid: 0x1234, pid: 0x5678 }, - HotplugEvent::Attached { busid: "2-1".into(), vid: 0x9abc, pid: 0xdef0 }, - ]; - let mut source = FakeHotplugSource::new(events.clone()); - - assert_eq!(source.poll(), events.get(0).cloned()); - assert_eq!(source.poll(), events.get(1).cloned()); - // After exhaustion, poll returns None. - assert_eq!(source.poll(), None); - } - - // ── HotplugMonitor ─────────────────────────────────────────── - - #[test] - fn test_monitor_receives_attach_event() { - let event = HotplugEvent::Attached { busid: "3-2".into(), vid: 0x046d, pid: 0xc261 }; - let source = FakeHotplugSource::new(vec![event.clone()]); - let monitor = HotplugMonitor::new(source); - - // Give the background thread time to poll and send. - std::thread::sleep(std::time::Duration::from_millis(50)); - - let received = monitor.try_recv(); - assert_eq!(received, Some(event)); - } - - #[test] - fn test_monitor_receives_detach_event() { - let cid = CorrelationId::now_v7(); - let event = HotplugEvent::Detached { busid: "1-1".into(), correlation_id: cid }; - let source = FakeHotplugSource::new(vec![event.clone()]); - let monitor = HotplugMonitor::new(source); - - std::thread::sleep(std::time::Duration::from_millis(50)); - - let received = monitor.try_recv(); - assert_eq!(received, Some(event)); - } - - #[test] - fn test_monitor_noop_source_never_emits() { - let source = NoopHotplugSource; - let monitor = HotplugMonitor::new(source); - - std::thread::sleep(std::time::Duration::from_millis(50)); - - assert_eq!(monitor.try_recv(), None); - } - - #[test] - fn test_monitor_recv_multiple_events_in_order() { - let events = vec![ - HotplugEvent::Attached { busid: "1-1".into(), vid: 0xaaaa, pid: 0xbbbb }, - HotplugEvent::Attached { busid: "1-2".into(), vid: 0xcccc, pid: 0xdddd }, - HotplugEvent::Detached { busid: "1-1".into(), correlation_id: CorrelationId::now_v7() }, - ]; - let source = FakeHotplugSource::new(events.clone()); - let monitor = HotplugMonitor::new(source); - - std::thread::sleep(std::time::Duration::from_millis(80)); - - assert_eq!(monitor.try_recv(), Some(events[0].clone())); - assert_eq!(monitor.try_recv(), Some(events[1].clone())); - assert_eq!(monitor.try_recv(), Some(events[2].clone())); - } - - #[test] - fn test_hotplug_events_are_send_sync() { - fn assert_send_sync() {} - assert_send_sync::(); - } - - #[test] - fn test_hotplug_event_clone() { - let event = HotplugEvent::Attached { busid: "3-2".into(), vid: 0x046d, pid: 0xc261 }; - let cloned = event.clone(); - assert_eq!(event, cloned); - } - - #[test] - fn test_hotplug_event_debug() { - let event = HotplugEvent::Attached { busid: "3-2".into(), vid: 0x046d, pid: 0xc261 }; - let debug_str = format!("{:?}", event); - assert!(debug_str.contains("Attached")); - assert!(debug_str.contains("3-2")); - } -} diff --git a/server/usbip-server/src/latency_hwm.rs b/server/usbip-server/src/latency_hwm.rs new file mode 100644 index 0000000..ae37451 --- /dev/null +++ b/server/usbip-server/src/latency_hwm.rs @@ -0,0 +1,170 @@ +//! Per-thread high-water-mark for URB latency. +//! +//! Tracks a thread-local peak and conditionally CAS-updates a shared +//! global peak (`Arc`). Broadcasts a `LatencySample` only +//! when a new global peak is established, eliminating nearly all +//! cross-thread CAS and broadcast operations. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +use tokio::sync::broadcast; + +use crate::api::LatencySample; + +/// Per-thread high-water-mark tracker for URB round-trip latency. +/// +/// Each URB loop owns one instance. On every completed URB the loop +/// calls [`observe`](Self::observe) with the elapsed microseconds. +/// The struct keeps a thread-local peak; only when that peak exceeds +/// the shared global peak does it attempt a CAS. On CAS success it +/// broadcasts a single [`LatencySample`] to the WebSocket channel. +pub(crate) struct HighWaterMark { + local_peak_us: u64, + global_peak: Arc, + busid: String, + latency_tx: broadcast::Sender, +} + +impl HighWaterMark { + pub(crate) fn new( + busid: String, + global_peak: Arc, + latency_tx: broadcast::Sender, + ) -> Self { + Self { local_peak_us: 0, global_peak, busid, latency_tx } + } + + /// Record an observed URB round-trip latency in microseconds. + /// + /// If this exceeds the thread-local peak, update it. If the new + /// local peak exceeds the global peak, attempt a CAS. On CAS + /// success, broadcast a [`LatencySample`]. On CAS failure + /// (another thread won), snap the local peak to the new global + /// value. + pub(crate) fn observe(&mut self, elapsed_us: u64, seqnum: u32) { + if elapsed_us <= self.local_peak_us { + return; + } + self.local_peak_us = elapsed_us; + + let current_global = self.global_peak.load(Ordering::Relaxed); + if elapsed_us <= current_global { + return; + } + + match self.global_peak.compare_exchange_weak( + current_global, + elapsed_us, + Ordering::AcqRel, + Ordering::Relaxed, + ) { + Ok(_) => { + let sample = LatencySample { + latency_us: elapsed_us, + device: self.busid.clone(), + seqnum: seqnum as u64, + }; + let _ = self.latency_tx.send(sample); + }, + Err(actual) => { + // Another thread won — sync local to global. + self.local_peak_us = actual; + }, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn local_peak_tracking() { + let global = Arc::new(AtomicU64::new(0)); + let (tx, _rx) = broadcast::channel(16); + let mut hwm = HighWaterMark::new("1-1".into(), Arc::clone(&global), tx); + + hwm.observe(100, 1); + assert_eq!(hwm.local_peak_us, 100); + + // Below peak -- no update. + hwm.observe(50, 2); + assert_eq!(hwm.local_peak_us, 100); + + // Above local peak -- local updates and CAS fires. + hwm.observe(200, 3); + assert_eq!(hwm.local_peak_us, 200); + assert_eq!(global.load(Ordering::Relaxed), 200); + } + + #[test] + fn cas_broadcasts_once_on_new_peak() { + let global = Arc::new(AtomicU64::new(0)); + let (tx, mut rx) = broadcast::channel(16); + let mut hwm = HighWaterMark::new("2-2".into(), Arc::clone(&global), tx); + + // First observation sets global peak and broadcasts. + hwm.observe(500, 10); + assert_eq!(rx.try_recv().unwrap().latency_us, 500); + + // Same value -- local peak matches, no broadcast. + hwm.observe(500, 11); + assert!(rx.try_recv().is_err()); + + // Lower -- no broadcast. + hwm.observe(300, 12); + assert!(rx.try_recv().is_err()); + + // Higher -- broadcasts once. + hwm.observe(600, 13); + let sample = rx.try_recv().unwrap(); + assert_eq!(sample.latency_us, 600); + assert_eq!(sample.device, "2-2"); + assert_eq!(sample.seqnum, 13); + assert!(rx.try_recv().is_err()); + } + + #[test] + fn below_global_peak_no_cas() { + let global = Arc::new(AtomicU64::new(1000)); + let (tx, mut rx) = broadcast::channel(16); + let mut hwm = HighWaterMark::new("3-3".into(), Arc::clone(&global), tx); + + // Below global -- no broadcast. + hwm.observe(500, 1); + assert!(rx.try_recv().is_err()); + assert_eq!(global.load(Ordering::Relaxed), 1000); + } + + #[test] + fn concurrent_threads_only_new_peak_broadcasts() { + let global = Arc::new(AtomicU64::new(0)); + let (tx, mut rx) = broadcast::channel(1024); + let mut handles = vec![]; + + for thread_id in 0..4u32 { + let g = Arc::clone(&global); + let t = tx.clone(); + handles.push(std::thread::spawn(move || { + let mut hwm = HighWaterMark::new(format!("t{thread_id}"), g, t); + for i in 0..1000u32 { + let latency = 100 + thread_id * 10; + hwm.observe(latency as u64, i); + } + })); + } + for h in handles { + h.join().unwrap(); + } + + // With 4 threads at distinct latencies (100, 110, 120, 130), + // at most 4 broadcasts fire (one per new peak value). + let mut count = 0; + while rx.try_recv().is_ok() { + count += 1; + } + assert!(count <= 4, "expected at most 4 broadcasts, got {count}"); + assert!(count >= 1, "expected at least 1 broadcast"); + } +} diff --git a/server/usbip-server/src/lib.rs b/server/usbip-server/src/lib.rs index 2995974..adea6ea 100644 --- a/server/usbip-server/src/lib.rs +++ b/server/usbip-server/src/lib.rs @@ -3,9 +3,9 @@ pub mod bandwidth; pub mod batcher; pub mod crypto_stream; pub mod discovery; -pub mod hotplug; #[cfg(target_os = "macos")] pub mod iokit_backend; +mod latency_hwm; pub mod metrics; pub mod server; pub mod usb; @@ -14,7 +14,6 @@ pub mod usb_backend; pub use api::AppState; pub use bandwidth::BandwidthLimit; pub use batcher::UrbBatcher; -pub use hotplug::{HotplugEvent, HotplugMonitor, HotplugSource, NoopHotplugSource}; pub use server::{Server, ServerConfig}; pub use usb::{UrbResult, UsbDeviceManager}; pub use usb_backend::{FakeBackend, UrbTransferResult, UsbBackend}; diff --git a/server/usbip-server/src/metrics.rs b/server/usbip-server/src/metrics.rs index 58b7bfc..29bd5aa 100644 --- a/server/usbip-server/src/metrics.rs +++ b/server/usbip-server/src/metrics.rs @@ -7,36 +7,10 @@ use std::sync::LazyLock; use axum::{extract::State, http::StatusCode, response::IntoResponse, routing::get, Router}; -use prometheus::{ - register_int_counter, register_int_gauge, Encoder, IntCounter, IntGauge, TextEncoder, -}; +use prometheus::{register_int_gauge, Encoder, IntGauge, TextEncoder}; // ── Metric definitions (lazily registered once) ───────────────────── -/// Number of devices currently exported (gauge). -pub static DEVICES_EXPORTED: LazyLock = LazyLock::new(|| { - register_int_gauge!("usbip_devices_exported", "Number of USB devices currently exported") - .expect("metric registration failed") -}); - -/// Number of active TCP client connections (gauge). -pub static CLIENTS_CONNECTED: LazyLock = LazyLock::new(|| { - register_int_gauge!("usbip_clients_connected", "Number of active TCP client connections") - .expect("metric registration failed") -}); - -/// Total number of URB submissions processed (counter). -pub static URB_SUBMIT_TOTAL: LazyLock = LazyLock::new(|| { - register_int_counter!("usbip_urb_submit_total", "Total number of URB submissions processed") - .expect("metric registration failed") -}); - -/// Total bytes transferred in URB payloads (counter). -pub static URB_BYTES_TOTAL: LazyLock = LazyLock::new(|| { - register_int_counter!("usbip_urb_bytes_total", "Total bytes transferred in URB payloads") - .expect("metric registration failed") -}); - /// Whether encryption is enabled; 1 = enabled, 0 = disabled (gauge). pub static ENCRYPTION_ENABLED: LazyLock = LazyLock::new(|| { register_int_gauge!( diff --git a/server/usbip-server/src/server.rs b/server/usbip-server/src/server.rs index dcf4a2c..1b231f9 100644 --- a/server/usbip-server/src/server.rs +++ b/server/usbip-server/src/server.rs @@ -15,11 +15,12 @@ use std::collections::HashMap; use std::net::SocketAddr; +use std::sync::atomic::AtomicU64; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::Mutex; +use tokio::sync::{broadcast, Mutex}; use tracing::{debug, error, info, info_span, warn}; use uuid::Uuid; use zerocopy::FromBytes; @@ -34,10 +35,12 @@ use usbip_core::protocol::{ use usbip_core::urb::UsbIpCmdSubmit; use crate::api; +use crate::api::LatencySample; use crate::bandwidth::BandwidthLimit; use crate::batcher::UrbBatcher; use crate::crypto_stream::Wire; use crate::discovery::{MdnsAdvertiser, MdnsBrowserImpl}; +use crate::latency_hwm; use crate::usb::UsbDeviceManager; use crate::usb_backend::UsbBackend; @@ -51,6 +54,12 @@ pub struct Server { pub mdns: Option, /// Server configuration. pub config: ServerConfig, + /// Shared high-water-mark peak for URB latency across all + /// connection tasks. + urb_latency_peak: Arc, + /// Broadcast channel for per-URB latency samples. Shared with + /// the WebSocket handler via `AppState`. + latency_tx: broadcast::Sender, } #[derive(Debug, Clone)] @@ -87,7 +96,14 @@ impl Server { let usb = UsbDeviceManager::new()?; let devices = usb.list_exportable_devices(&config.allowed_vid_pid); let mdns = MdnsAdvertiser::new(config.port, devices).ok(); - Ok(Self { usb: Arc::new(usb), exports: Arc::new(Mutex::new(HashMap::new())), mdns, config }) + Ok(Self { + usb: Arc::new(usb), + exports: Arc::new(Mutex::new(HashMap::new())), + mdns, + config, + urb_latency_peak: Arc::new(AtomicU64::new(0)), + latency_tx: api::new_latency_sender(), + }) } /// Create a server with a specific USB backend (for testing or non-libusb platforms). @@ -102,7 +118,14 @@ impl Server { let usb = UsbDeviceManager::with_backend(backend); let devices = usb.list_exportable_devices(&config.allowed_vid_pid); let mdns = MdnsAdvertiser::new(config.port, devices).ok(); - Ok(Self { usb: Arc::new(usb), exports: Arc::new(Mutex::new(HashMap::new())), mdns, config }) + Ok(Self { + usb: Arc::new(usb), + exports: Arc::new(Mutex::new(HashMap::new())), + mdns, + config, + urb_latency_peak: Arc::new(AtomicU64::new(0)), + latency_tx: api::new_latency_sender(), + }) } /// Run the server — listens forever. @@ -125,9 +148,21 @@ impl Server { let usb = self.usb.clone(); let exports = self.exports.clone(); let config = self.config.clone(); + let urb_latency_peak = Arc::clone(&self.urb_latency_peak); + let latency_tx = self.latency_tx.clone(); tokio::spawn(async move { - if let Err(e) = handle_client(stream, peer_addr, usb, exports, config).await { + if let Err(e) = handle_client( + stream, + peer_addr, + usb, + exports, + config, + urb_latency_peak, + latency_tx, + ) + .await + { error!("Client {} error: {}", peer_addr, e); } }); @@ -181,7 +216,7 @@ impl Server { cfg.encryption_enabled = self.config.encryption_enabled; cfg })), - latency_tx: api::new_latency_sender(), + latency_tx: self.latency_tx.clone(), } } @@ -222,9 +257,21 @@ impl Server { let usb = self.usb.clone(); let exports = self.exports.clone(); let config = self.config.clone(); + let urb_latency_peak = Arc::clone(&self.urb_latency_peak); + let latency_tx = self.latency_tx.clone(); tokio::spawn(async move { - if let Err(e) = handle_client(stream, peer_addr, usb, exports, config).await { + if let Err(e) = handle_client( + stream, + peer_addr, + usb, + exports, + config, + urb_latency_peak, + latency_tx, + ) + .await + { error!("Client {} error: {}", peer_addr, e); } }); @@ -242,6 +289,8 @@ pub async fn handle_client( usb: Arc, exports: Arc>>, config: ServerConfig, + urb_latency_peak: Arc, + latency_tx: broadcast::Sender, ) -> UsbIpResult<()> { let correlation_id = Uuid::now_v7(); let span = info_span!("handle_client", correlation_id = %correlation_id); @@ -264,8 +313,16 @@ pub async fn handle_client( match header.command.get() { OP_REQ_DEVLIST => handle_devlist(&mut stream, &usb).await?, OP_REQ_IMPORT => { - handle_import(stream, usb.clone(), &exports, config.encryption_enabled, peer_addr) - .await? + handle_import( + stream, + usb.clone(), + &exports, + config.encryption_enabled, + peer_addr, + urb_latency_peak, + latency_tx, + ) + .await? }, _ => { warn!("Unknown command: 0x{:04x}", header.command.get()); @@ -310,6 +367,8 @@ async fn handle_import( exports: &Mutex>, encryption_enabled: bool, peer_addr: SocketAddr, + urb_latency_peak: Arc, + latency_tx: broadcast::Sender, ) -> UsbIpResult<()> { // Read busid (32 bytes) let mut busid_buf = [0u8; 32]; @@ -376,7 +435,8 @@ async fn handle_import( }; // Enter URB forwarding loop - handle_urb_loop(&mut wire, usb.clone(), exports, busid, peer_addr).await + handle_urb_loop(&mut wire, usb.clone(), exports, busid, peer_addr, urb_latency_peak, latency_tx) + .await } /// Main URB forwarding loop after device import. @@ -386,6 +446,8 @@ async fn handle_urb_loop( exports: &Mutex>, busid: String, peer_addr: SocketAddr, + urb_latency_peak: Arc, + latency_tx: broadcast::Sender, ) -> UsbIpResult<()> { let correlation_id = Uuid::now_v7(); let span = @@ -393,6 +455,7 @@ async fn handle_urb_loop( let _guard = span.enter(); let mut batcher = UrbBatcher::new(); + let mut hwm = latency_hwm::HighWaterMark::new(busid.clone(), urb_latency_peak, latency_tx); loop { // Read a full USB/IP message — header + (variable) payload. @@ -434,7 +497,10 @@ async fn handle_urb_loop( &[] }; + let submit_start = std::time::Instant::now(); let result = usb.submit_urb(&busid, &cmd, data); + let elapsed_us = submit_start.elapsed().as_micros() as u64; + hwm.observe(elapsed_us, cmd.seqnum()); // Batch the reply — flush when full, non-sequential, or timed out. if batcher.push(&cmd, &result) { @@ -637,7 +703,9 @@ mod import_tests { let server = tokio::spawn(async move { let (stream, peer) = listener.accept().await.unwrap(); - handle_import(stream, usb, &exports, false, peer).await.unwrap(); + let peak = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let (tx, _rx) = tokio::sync::broadcast::channel(16); + handle_import(stream, usb, &exports, false, peer, peak, tx).await.unwrap(); }); // Client: send the 8-byte request header + 32-byte busid "9-9". diff --git a/server/usbip-server/src/usb.rs b/server/usbip-server/src/usb.rs index 2e0cc09..d68ff23 100644 --- a/server/usbip-server/src/usb.rs +++ b/server/usbip-server/src/usb.rs @@ -114,8 +114,10 @@ impl UsbDeviceManager { /// caller can always serialise a valid `USBIP_RET_SUBMIT` wire reply. pub fn submit_urb(&self, busid: &str, cmd: &UsbIpCmdSubmit, out_data: &[u8]) -> UrbResult { match self.execute_urb(busid, cmd, out_data) { - Ok(transfer) => { - UrbResult { status: transfer.status, actual_length: transfer.actual_length, data: transfer.data } + Ok(transfer) => UrbResult { + status: transfer.status, + actual_length: transfer.actual_length, + data: transfer.data, }, Err(e) => { let status = match e.kind() { diff --git a/server/usbip-server/tests/encryption_wire_test.rs b/server/usbip-server/tests/encryption_wire_test.rs index de80417..e673cde 100644 --- a/server/usbip-server/tests/encryption_wire_test.rs +++ b/server/usbip-server/tests/encryption_wire_test.rs @@ -8,6 +8,7 @@ //! (length 32) — never 0x00000111 (USB/IP version). use std::net::SocketAddr; +use std::sync::Arc; use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; @@ -59,7 +60,9 @@ async fn test_encrypted_stream_post_handshake_bytes_not_plaintext() { let (stream, peer) = listener.accept().await.unwrap(); let exports = std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())); - let _ = handle_client(stream, peer, usb_clone, exports, config).await; + let peak = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let (tx, _rx) = tokio::sync::broadcast::channel(16); + let _ = handle_client(stream, peer, usb_clone, exports, config, peak, tx).await; }); let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap(); diff --git a/server/usbip-server/tests/hotplug_integration.rs b/server/usbip-server/tests/hotplug_integration.rs deleted file mode 100644 index 9f3520b..0000000 --- a/server/usbip-server/tests/hotplug_integration.rs +++ /dev/null @@ -1,269 +0,0 @@ -//! Integration tests for hotplug detection integration with the server. -//! -//! Tests that the server responds correctly to USB device attach and detach -//! events: updating the export list, tearing down active imports, and -//! not crashing under any condition. - -use std::collections::HashMap; -use std::net::SocketAddr; -use std::sync::Arc; - -use tokio::sync::Mutex; -use usbip_core::error::CorrelationId; -use usbip_core::protocol::{UsbIpDeviceEntry, U16BE, U32BE}; - -use usbip_server::hotplug::{HotplugEvent, HotplugSource}; - -// ─── Helpers ──────────────────────────────────────────────────────────────── - -/// Build a `UsbIpDeviceEntry` with the given busid, VID, and PID. -fn device_entry(busid: &str, vid: u16, pid: u16) -> UsbIpDeviceEntry { - let busid_bytes = busid.as_bytes(); - let mut busid_arr = [0u8; 32]; - let copy_len = busid_bytes.len().min(31); - busid_arr[..copy_len].copy_from_slice(&busid_bytes[..copy_len]); - - // Build a path string. - let path_str = format!("/sys/bus/usb/devices/{}", busid); - let path_bytes = path_str.as_bytes(); - let mut path_arr = [0u8; 256]; - let copy_len = path_bytes.len().min(255); - path_arr[..copy_len].copy_from_slice(&path_bytes[..copy_len]); - - // Parse busnum and devnum from busid ("busnum-devnum"). - let parts: Vec<&str> = busid.split('-').collect(); - let busnum: u8 = parts.first().and_then(|s| s.parse().ok()).unwrap_or(0); - let devnum: u8 = parts.get(1).and_then(|s| s.parse().ok()).unwrap_or(0); - - UsbIpDeviceEntry { - path: path_arr, - busid: busid_arr, - busnum: U32BE::new(busnum as u32), - devnum: U32BE::new(devnum as u32), - speed: U32BE::new(3), - id_vendor: U16BE::new(vid), - id_product: U16BE::new(pid), - bcd_device: U16BE::new(0x0100), - b_device_class: 0, - b_device_sub_class: 0, - b_device_protocol: 0, - b_configuration_value: 1, - b_num_configurations: 1, - b_num_interfaces: 1, - } -} - -/// Minimal stand-in for the server's mutable export state. -fn make_exports( - entries: Vec<(&str, &str, u16, u16)>, -) -> Arc>> { - let map: HashMap = entries - .into_iter() - .map(|(busid, addr_str, vid, pid)| { - let addr: SocketAddr = addr_str.parse().unwrap(); - let entry = device_entry(busid, vid, pid); - (busid.to_string(), (addr, entry)) - }) - .collect(); - Arc::new(Mutex::new(map)) -} - -/// A fake hotplug source that yields a fixed sequence of events. -#[allow(dead_code)] -struct FakeHotplugSource { - events: Vec>, - index: usize, -} - -impl FakeHotplugSource { - #[allow(dead_code)] - fn new(events: Vec) -> Self { - Self { events: events.into_iter().map(Some).collect(), index: 0 } - } -} - -impl HotplugSource for FakeHotplugSource { - fn poll(&mut self) -> Option { - if self.index < self.events.len() { - let event = self.events[self.index].take(); - self.index += 1; - event - } else { - std::thread::sleep(std::time::Duration::from_millis(10)); - None - } - } -} - -// ─── Tests ───────────────────────────────────────────────────────────────── - -/// Simulate: attach event -> device should be added to exports. -#[test] -fn test_attach_event_adds_to_exports() { - let exports = make_exports(vec![]); - - let busid = "003-002"; - let vid = 0x046d; - let pid = 0xc261; - let entry = device_entry(busid, vid, pid); - - // Insert into exports (simulating what server does on attach). - let rt = tokio::runtime::Runtime::new().unwrap(); - rt.block_on(async { - let mut exports = exports.lock().await; - let addr: SocketAddr = "192.168.1.10:3240".parse().unwrap(); - exports.insert(busid.to_string(), (addr, entry)); - }); - - // Verify device is now in exports. - rt.block_on(async { - let exports = exports.lock().await; - let found = exports.get(busid); - assert!(found.is_some(), "Device should be in exports after attach"); - let (_, dev_entry) = found.unwrap(); - assert_eq!(dev_entry.id_vendor.get(), vid); - assert_eq!(dev_entry.id_product.get(), pid); - }); -} - -/// Simulate: detach event -> active import torn down. -#[test] -fn test_detach_event_removes_from_exports() { - let exports = make_exports(vec![("003-002", "192.168.1.10:3240", 0x046d, 0xc261)]); - - let busid = "003-002"; - - let rt = tokio::runtime::Runtime::new().unwrap(); - rt.block_on(async { - let mut exports = exports.lock().await; - let removed = exports.remove(busid); - assert!(removed.is_some(), "Device should be removed from exports on detach"); - assert!(!exports.contains_key(busid), "Busid should no longer be in exports"); - }); -} - -/// Simulate: detach event includes correlation ID linking to in-flight URB error. -#[test] -fn test_detach_event_correlation_id_links_to_urb_error() { - // This is a protocol test: when a device is detached mid-transfer, - // the server should emit a structured error with the same correlation ID - // that the detach event carries. This allows the client to correlate - // the in-flight URB failure with the device removal. - - let cid = CorrelationId::now_v7(); - - let event = HotplugEvent::Detached { busid: "003-002".into(), correlation_id: cid }; - - // Verify correlation_id is a UUIDv7. - assert_eq!(cid.as_bytes()[6] >> 4, 7, "CorrelationId must be UUIDv7"); - - // Verify we can extract the same correlation ID from the event. - match &event { - HotplugEvent::Detached { busid, correlation_id } => { - assert_eq!(busid, "003-002"); - assert_eq!(*correlation_id, cid); - }, - _ => panic!("expected Detached"), - } -} - -/// Simulate: detach and re-attach of the same busid. -#[test] -fn test_detach_then_attach_same_device() { - let exports = make_exports(vec![("003-002", "192.168.1.10:3240", 0x046d, 0xc261)]); - - let rt = tokio::runtime::Runtime::new().unwrap(); - - // Detach - { - rt.block_on(async { - let mut exports = exports.lock().await; - exports.remove("003-002"); - assert!(exports.is_empty()); - }); - } - - // Re-attach same busid (different device plugged into same port) - { - let entry = device_entry("003-002", 0x1234, 0x5678); - - rt.block_on(async { - let mut exports = exports.lock().await; - let addr: SocketAddr = "192.168.1.20:3240".parse().unwrap(); - exports.insert("003-002".to_string(), (addr, entry)); - }); - } - - // Verify the device is back with new VID:PID. - rt.block_on(async { - let exports = exports.lock().await; - let found = exports.get("003-002"); - assert!(found.is_some(), "Device should be back in exports after re-attach"); - let (_, dev_entry) = found.unwrap(); - assert_eq!(dev_entry.id_vendor.get(), 0x1234); - assert_eq!(dev_entry.id_product.get(), 0x5678); - }); -} - -/// Simulate: multiple attach events add multiple devices. -#[test] -fn test_multiple_attach_events() { - let exports = make_exports(vec![]); - - let rt = tokio::runtime::Runtime::new().unwrap(); - rt.block_on(async { - let mut exports = exports.lock().await; - let addr: SocketAddr = "192.168.1.10:3240".parse().unwrap(); - - let dev1 = device_entry("001-001", 0x046d, 0xc261); - exports.insert("001-001".to_string(), (addr, dev1)); - - let dev2 = device_entry("002-003", 0x1234, 0x5678); - exports.insert("002-003".to_string(), (addr, dev2)); - - assert_eq!(exports.len(), 2); - }); -} - -/// Simulate: detach of non-exported device is a no-op (no crash). -#[test] -fn test_detach_nonexistent_device_no_crash() { - let exports = make_exports(vec![]); - - let rt = tokio::runtime::Runtime::new().unwrap(); - rt.block_on(async { - let mut exports = exports.lock().await; - let removed = exports.remove("999-999"); - assert!(removed.is_none(), "Removing non-existent device should return None"); - }); -} - -/// Simulate: concurrent attach and detach (via serialised Mutex access). -#[test] -fn test_concurrent_attach_detach_does_not_crash() { - let exports = make_exports(vec![]); - let exports_clone = exports.clone(); - - let rt = tokio::runtime::Runtime::new().unwrap(); - - rt.block_on(async { - // Send an attach event. - let mut exports = exports.lock().await; - let entry = device_entry("003-002", 0x046d, 0xc261); - let addr: SocketAddr = "192.168.1.10:3240".parse().unwrap(); - exports.insert("003-002".to_string(), (addr, entry)); - }); - - rt.block_on(async { - // Simultaneous detach from another path. - let mut exports = exports_clone.lock().await; - exports.remove("003-002"); - assert!(exports.is_empty()); - }); - - // The server should not panic or hang. - rt.block_on(async { - let exports = exports.lock().await; - assert!(exports.is_empty()); - }); -} diff --git a/server/usbip-server/tests/server_wire_test.rs b/server/usbip-server/tests/server_wire_test.rs index 499ce96..455a275 100644 --- a/server/usbip-server/tests/server_wire_test.rs +++ b/server/usbip-server/tests/server_wire_test.rs @@ -5,6 +5,7 @@ //! localhost, and drops the client stream to shut down cleanly. use std::net::SocketAddr; +use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; @@ -59,7 +60,9 @@ async fn test_devlist_returns_fake_devices() { let (stream, peer) = listener.accept().await.unwrap(); let exports = std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())); - let _ = handle_client(stream, peer, usb_clone, exports, config).await; + let peak = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let (tx, _rx) = tokio::sync::broadcast::channel(16); + let _ = handle_client(stream, peer, usb_clone, exports, config, peak, tx).await; }); let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap(); @@ -115,7 +118,9 @@ async fn test_import_valid_device_returns_success() { let (stream, peer) = listener.accept().await.unwrap(); let exports = std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())); - let _ = handle_client(stream, peer, usb_clone, exports, config).await; + let peak = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let (tx, _rx) = tokio::sync::broadcast::channel(16); + let _ = handle_client(stream, peer, usb_clone, exports, config, peak, tx).await; }); let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap(); diff --git a/shared/usbip-core/Cargo.toml b/shared/usbip-core/Cargo.toml index bf1ed1d..52b1291 100644 --- a/shared/usbip-core/Cargo.toml +++ b/shared/usbip-core/Cargo.toml @@ -26,3 +26,7 @@ toml = "0.8" [[bench]] name = "urb_pool_bench" harness = false + +[[bench]] +name = "aead_inplace_bench" +harness = false diff --git a/shared/usbip-core/benches/aead_inplace_bench.rs b/shared/usbip-core/benches/aead_inplace_bench.rs new file mode 100644 index 0000000..b3eadcd --- /dev/null +++ b/shared/usbip-core/benches/aead_inplace_bench.rs @@ -0,0 +1,83 @@ +//! Micro-benchmark: in-place AEAD decrypt vs allocating decrypt. +//! +//! Compares the latency of `decrypt_in_place` (zero-copy, mutates the +//! read buffer) against `decrypt` (allocates a fresh `Vec`) on a +//! representative 512-byte USB/IP packet. +//! +//! Run with: cargo bench -p usbip-core --bench aead_inplace_bench + +use std::time::Duration; + +use criterion::{black_box, criterion_group, criterion_main, Criterion}; +use usbip_core::crypto; + +/// Derive a deterministic AES-256-GCM session key from a fixed shared secret. +fn fixed_key() -> ring::aead::LessSafeKey { + let shared = [0x42u8; 32]; + crypto::derive_session_key(&shared).expect("key derivation must succeed") +} + +/// Encrypt a 512-byte plaintext once; returns the wire-format ciphertext. +fn encrypt_512(key: &ring::aead::LessSafeKey) -> Vec { + let plaintext = vec![0xABu8; 512]; + crypto::encrypt_message(key, &plaintext).expect("encryption must succeed") +} + +fn bench_decrypt_allocating(c: &mut Criterion) { + let key = fixed_key(); + let wire = encrypt_512(&key); + + c.bench_function("aead/decrypt_alloc_512", |b| { + b.iter(|| { + let pt = crypto::decrypt(black_box(&key), black_box(&wire)).unwrap(); + black_box(pt.len()); + }) + }); +} + +fn bench_decrypt_in_place(c: &mut Criterion) { + let key = fixed_key(); + let wire = encrypt_512(&key); + + c.bench_function("aead/decrypt_in_place_512", |b| { + b.iter(|| { + let mut buf = wire.clone(); + let pt = crypto::decrypt_in_place(black_box(&key), &mut buf).unwrap(); + black_box(pt.len()); + }) + }); +} + +fn bench_decrypt_byte_equality(c: &mut Criterion) { + let key = fixed_key(); + let wire = encrypt_512(&key); + + c.bench_function("aead/decrypt_byte_equality_check", |b| { + b.iter(|| { + // Allocating path + let wire_copy = wire.clone(); + let pt_alloc = crypto::decrypt(&key, &wire_copy).unwrap(); + + // In-place path + let mut buf = wire.clone(); + let pt_inplace = crypto::decrypt_in_place(&key, &mut buf).unwrap(); + + assert_eq!(pt_alloc, pt_inplace); + black_box(pt_alloc.len()); + }) + }); +} + +criterion_group! { + name = aead_benches; + config = Criterion::default() + .measurement_time(Duration::from_secs(3)) + .warm_up_time(Duration::from_secs(1)) + .sample_size(50); + targets = + bench_decrypt_allocating, + bench_decrypt_in_place, + bench_decrypt_byte_equality, +} + +criterion_main!(aead_benches); diff --git a/shared/usbip-core/src/crypto.rs b/shared/usbip-core/src/crypto.rs index 4f0bfa4..630fb8f 100644 --- a/shared/usbip-core/src/crypto.rs +++ b/shared/usbip-core/src/crypto.rs @@ -151,6 +151,10 @@ pub fn encrypt_with_key_bytes(key_bytes: &[u8], plaintext: &[u8]) -> CryptoResul /// Decrypt ciphertext produced by `encrypt_message()` / `encrypt_with_nonce()`. /// /// Wire format: `[4-byte nonce_len (=12)][12-byte nonce][ciphertext || 16-byte GCM tag]` +/// +/// Audit (issue #34): allocates a fresh `Vec` for the plaintext. +/// Call sites: `decrypt_with_key_bytes` (JNI bridge — must allocate for FFI), +/// unit tests only. Production server path uses `decrypt_in_place` instead. pub fn decrypt(key: &LessSafeKey, wire_data: &[u8]) -> CryptoResult> { if wire_data.len() < 4 + 12 + 16 { // minimum: 4-byte len + 12-byte nonce + 16-byte tag @@ -185,6 +189,10 @@ pub fn decrypt(key: &LessSafeKey, wire_data: &[u8]) -> CryptoResult> { /// (which `open_in_place` zeroes on success), and a borrowed slice of the /// remaining plaintext is returned. No reallocation occurs; the buffer's /// `capacity()` is preserved across the call. +/// +/// Audit (issue #34): in-place — mutates the caller's buffer, zero copies. +/// Call sites: `server/crypto_stream.rs:146` (CryptoStream::read_message, the +/// sole production decrypt path), unit test `test_decrypt_in_place_roundtrip`. pub fn decrypt_in_place<'a>( key: &'a LessSafeKey, buf: &'a mut Vec, diff --git a/shared/usbip-core/src/discovery_txt.rs b/shared/usbip-core/src/discovery_txt.rs new file mode 100644 index 0000000..ebd6cad --- /dev/null +++ b/shared/usbip-core/src/discovery_txt.rs @@ -0,0 +1,253 @@ +//! Encode and decode the `devices` TXT value used in mDNS advertisements. +//! +//! The wire format is a comma-separated list of `vid=0xVVVV,pid=0xPPPP,bus=B-B,n=NAME` +//! tuples, where `vid` and `pid` are lowercase hex, `bus` is a raw USB bus-id string +//! (e.g. `1-1`), and `name` is a human-readable device name (today the +//! `vid:pid` hex pair). +//! +//! This module is the **single seam** for TXT serialization — both the server +//! (advertise) and the client (browse) use it. + +use crate::error::{ErrorKind, UsbIpResult}; +use crate::protocol::UsbIpDeviceEntry; + +/// A device entry decoded from the mDNS `devices` TXT key. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct DiscoveredDevice { + pub vid: u16, + pub pid: u16, + pub bus: String, + pub name: String, +} + +/// Encode a slice of device entries into the `devices` TXT value. +/// +/// Format: comma-separated `vid=0xVVVV,pid=0xPPPP,bus=B-B,n=NAME` tuples. +/// Returns an empty string for an empty input. +pub fn encode_devices_txt(devices: &[UsbIpDeviceEntry]) -> String { + devices + .iter() + .map(|d| { + format!( + "vid=0x{:04x},pid=0x{:04x},bus={},n={:04x}:{:04x}", + d.vid(), + d.pid(), + d.busid_str(), + d.vid(), + d.pid() + ) + }) + .collect::>() + .join(",") +} + +/// Parse a `devices` TXT value back into a list of [`DiscoveredDevice`]. +/// +/// The wire format uses commas as both field separators (within a device +/// tuple) and tuple separators (between devices). Each device consists +/// of exactly 4 comma-separated fields in order: `vid=0x...`, +/// `pid=0x...`, `bus=...`, `n=...`. The parser groups every 4 +/// consecutive comma-separated tokens into one device. +/// +/// Returns an error if the token count is not a multiple of 4 or any +/// required field is missing or malformed. +pub fn decode_devices_txt(txt: &str) -> UsbIpResult> { + if txt.is_empty() { + return Ok(Vec::new()); + } + + let tokens: Vec<&str> = txt.split(',').collect(); + if tokens.is_empty() { + return Ok(Vec::new()); + } + if !tokens.len().is_multiple_of(4) { + return Err(ErrorKind::InvalidMessage(format!( + "device TXT token count {} is not a multiple of 4", + tokens.len() + )) + .into()); + } + + let mut devices = Vec::with_capacity(tokens.len() / 4); + + for chunk in tokens.chunks(4) { + let vid = parse_vid_pid_field(chunk[0], "vid")?; + let pid = parse_vid_pid_field(chunk[1], "pid")?; + let bus = parse_kv_field(chunk[2], "bus")?; + let name = parse_kv_field(chunk[3], "n")?; + devices.push(DiscoveredDevice { vid, pid, bus, name }); + } + + Ok(devices) +} + +/// Parse a `vid=0x...` or `pid=0x...` field, returning the u16 value. +fn parse_vid_pid_field(field: &str, expected_key: &str) -> UsbIpResult { + let prefix = format!("{expected_key}=0x"); + let val = field.strip_prefix(&prefix).ok_or_else(|| { + ErrorKind::InvalidMessage(format!("expected {expected_key}=0x..., got: {field}")) + })?; + u16::from_str_radix(val, 16).map_err(|_| { + { ErrorKind::InvalidMessage(format!("invalid hex in {expected_key}: {field}")) }.into() + }) +} + +/// Parse a `key=value` field, returning the value string. +fn parse_kv_field(field: &str, expected_key: &str) -> UsbIpResult { + let prefix = format!("{expected_key}="); + let val = field.strip_prefix(&prefix).ok_or_else(|| { + ErrorKind::InvalidMessage(format!("expected {expected_key}=..., got: {field}")) + })?; + Ok(val.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::U16BE; + use crate::protocol::U32BE; + + /// Build a minimal `UsbIpDeviceEntry` for testing. + fn make_test_entry(busid: &str, vid: u16, pid: u16) -> UsbIpDeviceEntry { + let mut entry = UsbIpDeviceEntry { + path: [0u8; 256], + busid: [0u8; 32], + busnum: U32BE::new(1), + devnum: U32BE::new(1), + speed: U32BE::new(3), + id_vendor: U16BE::new(vid), + id_product: U16BE::new(pid), + bcd_device: U16BE::new(0x0100), + b_device_class: 0, + b_device_sub_class: 0, + b_device_protocol: 0, + b_configuration_value: 1, + b_num_configurations: 1, + b_num_interfaces: 1, + }; + let busid_bytes = busid.as_bytes(); + let copy_len = busid_bytes.len().min(31); + entry.busid[..copy_len].copy_from_slice(&busid_bytes[..copy_len]); + entry + } + + // -- encode tests (migrated from server discovery.rs) -- + + #[test] + fn encode_empty() { + assert_eq!(encode_devices_txt(&[]), ""); + } + + #[test] + fn encode_single() { + let devices = vec![make_test_entry("1-1", 0x1234, 0x5678)]; + assert_eq!(encode_devices_txt(&devices), "vid=0x1234,pid=0x5678,bus=1-1,n=1234:5678"); + } + + #[test] + fn encode_multi_preserves_order() { + let devices = + vec![make_test_entry("1-1", 0x046d, 0xc261), make_test_entry("1-2", 0x8087, 0x0024)]; + assert_eq!( + encode_devices_txt(&devices), + "vid=0x046d,pid=0xc261,bus=1-1,n=046d:c261,vid=0x8087,pid=0x0024,bus=1-2,n=8087:0024" + ); + } + + // -- decode tests -- + + #[test] + fn decode_empty_string() { + assert_eq!(decode_devices_txt("").unwrap(), vec![]); + } + + #[test] + fn decode_single() { + let result = decode_devices_txt("vid=0x1234,pid=0x5678,bus=1-1,n=1234:5678").unwrap(); + assert_eq!(result.len(), 1); + assert_eq!( + result[0], + DiscoveredDevice { + vid: 0x1234, + pid: 0x5678, + bus: "1-1".into(), + name: "1234:5678".into() + } + ); + } + + #[test] + fn decode_multi() { + let txt = + "vid=0x046d,pid=0xc261,bus=1-1,n=046d:c261,vid=0x8087,pid=0x0024,bus=1-2,n=8087:0024"; + let result = decode_devices_txt(txt).unwrap(); + assert_eq!(result.len(), 2); + assert_eq!(result[0].vid, 0x046d); + assert_eq!(result[0].pid, 0xc261); + assert_eq!(result[0].bus, "1-1"); + assert_eq!(result[1].vid, 0x8087); + assert_eq!(result[1].pid, 0x0024); + assert_eq!(result[1].bus, "1-2"); + } + + // -- round-trip tests -- + + #[test] + fn round_trip_single() { + let devices = vec![make_test_entry("1-1", 0x1234, 0x5678)]; + let encoded = encode_devices_txt(&devices); + let decoded = decode_devices_txt(&encoded).unwrap(); + assert_eq!(decoded.len(), 1); + assert_eq!(decoded[0].vid, 0x1234); + assert_eq!(decoded[0].pid, 0x5678); + assert_eq!(decoded[0].bus, "1-1"); + assert_eq!(decoded[0].name, "1234:5678"); + } + + #[test] + fn round_trip_multi() { + let devices = vec![ + make_test_entry("1-1", 0x046d, 0xc261), + make_test_entry("1-2", 0x8087, 0x0024), + make_test_entry("3-4", 0x0aaa, 0x0bbb), + ]; + let encoded = encode_devices_txt(&devices); + let decoded = decode_devices_txt(&encoded).unwrap(); + assert_eq!(decoded.len(), 3); + for (i, dev) in devices.iter().enumerate() { + assert_eq!(decoded[i].vid, dev.vid()); + assert_eq!(decoded[i].pid, dev.pid()); + assert_eq!(decoded[i].bus, dev.busid_str()); + } + } + + // -- malformed input tests -- + + #[test] + fn decode_malformed_missing_field() { + let result = decode_devices_txt("vid=0x1234,pid=0x5678"); + assert!(result.is_err()); + } + + #[test] + fn decode_malformed_bad_hex() { + let result = decode_devices_txt("vid=0xZZZZ,pid=0x5678,bus=1-1,n=bad"); + assert!(result.is_err()); + } + + #[test] + fn decode_unknown_field_only() { + // Segment has no vid/pid/bus/n fields -- should fail + let result = decode_devices_txt("foo=bar"); + assert!(result.is_err()); + } + + #[test] + fn default_discovered_device() { + let d = DiscoveredDevice::default(); + assert_eq!(d.vid, 0); + assert_eq!(d.pid, 0); + assert_eq!(d.bus, ""); + assert_eq!(d.name, ""); + } +} diff --git a/shared/usbip-core/src/lib.rs b/shared/usbip-core/src/lib.rs index 4cd9450..76f3387 100644 --- a/shared/usbip-core/src/lib.rs +++ b/shared/usbip-core/src/lib.rs @@ -14,6 +14,7 @@ pub mod crypto; pub mod descriptor; +pub mod discovery_txt; pub mod error; pub mod pool; pub mod protocol; @@ -67,6 +68,10 @@ pub use reply::serialize_reply; pub use reply::serialize_reply_into; pub use reply::serialize_ret_submit; +pub use discovery_txt::decode_devices_txt; +pub use discovery_txt::encode_devices_txt; +pub use discovery_txt::DiscoveredDevice; + /// Default USB/IP TCP port (IANA-registered). pub const USBIP_PORT: u16 = 3240;