diff --git a/src/dialdbg/main.rs b/src/dialdbg/main.rs index a0ddc7b..4204ba3 100644 --- a/src/dialdbg/main.rs +++ b/src/dialdbg/main.rs @@ -400,5 +400,8 @@ pub(crate) async fn main_inner(args: Args) -> Result<()> { #[tokio::main] async fn main() -> Result<()> { + // dialdbg is a diagnostic tool; its dials are synthetic (and often use force_p2p/force_relay), + // so keep them out of the best-effort dial connection reports sent to app (see rpc::dial_report). + std::env::set_var("VIAM_DISABLE_DIAL_REPORTING", "1"); main_inner(Args::parse()).await } diff --git a/src/rpc/dial.rs b/src/rpc/dial.rs index e54c572..6ff396e 100644 --- a/src/rpc/dial.rs +++ b/src/rpc/dial.rs @@ -1,5 +1,6 @@ use super::{ client_channel::*, + dial_report::{self, StageTracker}, log_prefixes, webrtc::{webrtc_action_with_timeout, Options}, }; @@ -13,7 +14,8 @@ use crate::gen::proto::rpc::webrtc::v1::{ OptionalWebRtcConfigRequest, OptionalWebRtcConfigResponse, }; use crate::gen::proto::rpc::webrtc::v1::{ - CallRequest, IceCandidate, Metadata, RequestHeaders, Strings, + CallRequest, ConnectionSignalingPath, DialStage, IceCandidate, Metadata, + ReportConnectionMetadataRequest, RequestHeaders, Strings, }; use crate::rpc::webrtc; use ::http::header::HeaderName; @@ -725,6 +727,9 @@ impl DialBuilder { original_uri, self.config.signaling_server_override.as_deref(), ); + // `uri` is the resolved signaling server; classify how this WebRTC dial is signaled so a + // connection report can be delivered (only cloud-signaled dials report). + let signaling_path = dial_report::classify_signaling_path(uri.host().unwrap_or_default()); let domain = uri2.authority().to_owned().unwrap().as_str(); let mdns_uri = mdns_uri.and_then(|p| Uri::from_parts(p).ok()); @@ -780,7 +785,23 @@ impl DialBuilder { log::debug!("{}", log_prefixes::DIALED_GRPC); Ok(ViamChannel::Direct(channel.clone())) } else { - match maybe_connect_via_webrtc(uri, intercepted_channel.clone(), webrtc_options).await { + // A cloud-signaled dial reports over its own channel. A locally-signaled dial would + // report to prod app reusing credentials, but this (WithoutCredentials) dial has none, + // so it is dropped (matches Go's fixUpReportDialOpts). + let report_target = match signaling_path { + ConnectionSignalingPath::CloudSignaled => { + Some(ReportTarget::Cloud(intercepted_channel.clone())) + } + _ => None, + }; + match maybe_connect_via_webrtc( + uri, + intercepted_channel.clone(), + webrtc_options, + report_target, + ) + .await + { Ok(webrtc_channel) => Ok(ViamChannel::WebRTC(webrtc_channel)), Err(e) => { log::error!("error connecting via webrtc: {e}. Attempting to connect directly"); @@ -916,6 +937,18 @@ impl DialBuilder { original_uri.clone(), self.config.signaling_server_override.as_deref(), ); + // `uri_for_auth` is the resolved signaling server; classify how this WebRTC dial is + // signaled so a connection report can be delivered. + let signaling_path = + dial_report::classify_signaling_path(uri_for_auth.host().unwrap_or_default()); + // Capture credentials for the local-dial report case before they are consumed by auth + // below; a locally-signaled WebRTC dial reports to prod app reusing these (see + // ReportTarget::Local). + let report_creds = self + .config + .credentials + .as_ref() + .map(|c| (c.credentials.clone(), c.entity.clone())); let mdns_uri = mdns_uri.and_then(|p| Uri::from_parts(p).ok()); let attempting_mdns = mdns_uri.is_some(); @@ -982,7 +1015,30 @@ impl DialBuilder { log::debug!("Connected via gRPC"); Ok(ViamChannel::DirectPreAuthorized(channel)) } else { - match maybe_connect_via_webrtc(original_uri, channel.clone(), webrtc_options).await { + // A cloud-signaled dial reports over its own authenticated channel; a locally-signaled + // dial reconstructs an app connection reusing these credentials (Go's + // fixUpReportDialOpts). + let report_target = match signaling_path { + ConnectionSignalingPath::CloudSignaled => { + Some(ReportTarget::Cloud(channel.clone())) + } + ConnectionSignalingPath::Local => { + report_creds.map(|(creds, entity)| ReportTarget::Local { + creds, + entity: entity.unwrap_or_else(|| domain.clone()), + host: domain.clone(), + }) + } + _ => None, + }; + match maybe_connect_via_webrtc( + original_uri, + channel.clone(), + webrtc_options, + report_target, + ) + .await + { Ok(webrtc_channel) => Ok(ViamChannel::WebRTC(webrtc_channel)), Err(e) => { log::error!( @@ -1148,12 +1204,146 @@ impl fmt::Display for CallerUpdateStats { } } +/// The prod app signaling server, used to deliver reports for locally-signaled dials (Go's +/// fixUpReportDialOpts redirects such dials to prod app). Like Go, this is hard-coded to prod, so a +/// staging robot dialed by a local address does not route its report to app.viam.dev. +const APP_SIGNALING_ADDRESS_FOR_REPORT: &str = "app.viam.com:443"; + +/// How to deliver a WebRTC dial's connection report to the app signaling server (only app +/// implements ReportConnectionMetadata). Mirrors goutils#583's per-dial app-dial options. +enum ReportTarget { + /// The dial was cloud-signaled (routed through app); deliver over that same authenticated, + /// rpc-host-stamped channel. + Cloud(AddAuthorization>), + /// The dial was signaled through a local signaling server; reconstruct an authenticated + /// connection to prod app, reusing the dial's credentials, and deliver there (Go's + /// fixUpReportDialOpts). Only a cloud-managed robot dialed by a local address can authenticate + /// to app; for anything else the report RPC fails harmlessly. + Local { + creds: Credentials, + entity: String, + host: String, + }, +} + +impl ReportTarget { + fn signaling_path(&self) -> ConnectionSignalingPath { + match self { + ReportTarget::Cloud(_) => ConnectionSignalingPath::CloudSignaled, + ReportTarget::Local { .. } => ConnectionSignalingPath::Local, + } + } +} + +/// Reconstructs an authenticated connection to prod app to deliver a connection report for a +/// locally-signaled dial, reusing the dial's credentials (Go's fixUpReportDialOpts). The rpc-host +/// is the robot's host so app can attribute the report. app is always TLS. +async fn connect_app_signaling_for_report( + creds: Credentials, + entity: String, + host: String, +) -> Result>> { + let uri = Uri::from_parts(uri_parts_with_defaults(APP_SIGNALING_ADDRESS_FOR_REPORT))?; + let mut channel = Channel::builder(uri).connect().await?; + let token = get_auth_token(&mut channel, creds, entity).await?; + Ok(ServiceBuilder::new() + .layer(AddAuthorizationLayer::bearer(&token)) + .layer(SetRequestHeaderLayer::overriding( + HeaderName::from_static("rpc-host"), + HeaderValue::from_str(&host)?, + )) + .service(channel)) +} + +/// Delivers a connection report to the target, reconstructing an app connection first for a +/// locally-signaled dial. Best-effort; failures are logged at debug. +async fn deliver_report(target: ReportTarget, request: ReportConnectionMetadataRequest) { + match target { + ReportTarget::Cloud(channel) => dial_report::send_dial_report(channel, request).await, + ReportTarget::Local { + creds, + entity, + host, + } => { + let reconstruct = connect_app_signaling_for_report(creds, entity, host); + match tokio::time::timeout(Duration::from_secs(5), reconstruct).await { + Ok(Ok(channel)) => dial_report::send_dial_report(channel, request).await, + Ok(Err(e)) => { + log::debug!("failed to connect to app to report local dial metadata: {e:#}") + } + Err(_) => log::debug!("timed out connecting to app to report local dial metadata"), + } + } + } +} + +/// Attempts a WebRTC dial and, on completion (success or failure), best-effort reports the dial's +/// connection metadata to the app signaling server (a port of goutils#583). Reporting never adds +/// latency to or fails the dial. `report_target` is None when the dial should not be reported +/// (e.g. a locally-signaled dial with no credentials to authenticate to app with). async fn maybe_connect_via_webrtc( uri: Uri, channel: AddAuthorization>, webrtc_options: Option, + report_target: Option, +) -> Result> { + let dial_start = Instant::now(); + let stage_tracker = Arc::new(StageTracker::new()); + + let result = + maybe_connect_via_webrtc_inner(uri, channel, webrtc_options, stage_tracker.clone()).await; + + // Reporting is disabled under cfg(test). + if let Some(report_target) = report_target.filter(|_| dial_report::dial_reporting_enabled()) { + let duration_ms = dial_start.elapsed().as_millis().min(u32::MAX as u128) as u32; + let reached_stage = stage_tracker.reached(); + let dial_succeeded = result.is_ok(); + let signaling_path = report_target.signaling_path(); + + if dial_report::should_deliver(reached_stage, dial_succeeded, signaling_path) { + // On success, classify the selected ICE candidate pair from the live peer connection; + // a failed dial has no peer connection to inspect. + let peer_connection = result + .as_ref() + .ok() + .map(|client_channel| client_channel.base_channel.peer_connection.clone()); + let failure_code = result.as_ref().err().map_or(0, dial_report::failure_code); + + // Detached so reporting (gathering stats, reconnecting to app, and the RPC) never adds + // latency to the dial, and a cancelled dial still reports. + tokio::spawn(async move { + let (local, remote) = + dial_report::classify_connection(peer_connection.as_deref()).await; + let request = ReportConnectionMetadataRequest { + local: Some(local), + remote: Some(remote), + reached_stage, + duration_ms, + signaling_path: signaling_path as i32, + failure_code, + }; + deliver_report(report_target, request).await; + }); + } + } + + result +} + +async fn maybe_connect_via_webrtc_inner( + uri: Uri, + channel: AddAuthorization>, + webrtc_options: Option, + stage_tracker: Arc, ) -> Result> { let webrtc_options = webrtc_options.unwrap_or_else(|| Options::infer_from_uri(uri.clone())); + + // The channel was eagerly connected by create_channel (it uses `.connect().await`) before this + // function is called, so the signaling connection is already established on entry. Matching Go, + // set SIGNALING_CONNECTED here so a config-RPC failure against a reachable server reports + // SIGNALING_CONNECTED rather than UNSPECIFIED. + stage_tracker.advance(DialStage::SignalingConnected); + let mut signaling_client = SignalingServiceClient::new(channel.clone()); let response = match signaling_client .optional_web_rtc_config(OptionalWebRtcConfigRequest::default()) @@ -1170,6 +1360,7 @@ async fn maybe_connect_via_webrtc( }; let optional_config = response.into_inner().config; + stage_tracker.advance(DialStage::ConfigFetched); if webrtc_options.force_relay && webrtc_options.force_p2p { log::warn!( @@ -1210,8 +1401,12 @@ async fn maybe_connect_via_webrtc( log::debug!("TURN filter options set: turn_uri={uri:?}"); } - let (peer_connection, data_channel) = - webrtc::new_peer_connection_for_client(config, webrtc_options.disable_trickle_ice).await?; + let (peer_connection, data_channel) = webrtc::new_peer_connection_for_client( + config, + webrtc_options.disable_trickle_ice, + stage_tracker.clone(), + ) + .await?; let sent_done_or_error = Arc::new(AtomicBool::new(false)); let uuid_lock = Arc::new(RwLock::new("".to_string())); @@ -1246,10 +1441,17 @@ async fn maybe_connect_via_webrtc( let caller_update_stats = caller_update_stats.clone(); let caller_update_stats2 = caller_update_stats.clone(); + let stage_tracker_ice = stage_tracker.clone(); peer_connection.on_ice_connection_state_change(Box::new( move |state: RTCIceConnectionState| { let caller_update_stats = caller_update_stats.clone(); + let stage_tracker_ice = stage_tracker_ice.clone(); Box::pin(async move { + if state == RTCIceConnectionState::Connected + || state == RTCIceConnectionState::Completed + { + stage_tracker_ice.advance(DialStage::IceConnected); + } if state == RTCIceConnectionState::Completed { let caller_update_stats_inner = caller_update_stats.lock().unwrap(); log::debug!("{}", caller_update_stats_inner); @@ -1390,9 +1592,12 @@ async fn maybe_connect_via_webrtc( let client_channel_for_ice_gathering_thread = Arc::downgrade(&client_channel); let mut signaling_client = SignalingServiceClient::new(channel.clone()); let mut call_client = signaling_client.call(call_request).await?.into_inner(); + // The Call was accepted: the SDP offer has been sent to the signaling server. + stage_tracker.advance(DialStage::OfferSent); let channel2 = channel.clone(); let sent_done_or_error2 = sent_done_or_error.clone(); + let stage_tracker_task = stage_tracker.clone(); tokio::spawn(async move { let uuid = uuid_for_ice_gathering_thread; let client_channel = client_channel_for_ice_gathering_thread; @@ -1476,6 +1681,8 @@ async fn maybe_connect_via_webrtc( break; } } + // The answerer's SDP answer was received and applied. + stage_tracker_task.advance(DialStage::AnswerReceived); let _ = remote_description_set_s.send_replace(Some(())); if webrtc_options.disable_trickle_ice { send_done_once(sent_done.clone(), &response.uuid, channel2.clone()).await; @@ -1548,6 +1755,8 @@ async fn maybe_connect_via_webrtc( } exchange_done.store(true, Ordering::Release); + // The data channel is open: the connection is fully ready. This is success. + stage_tracker.advance(DialStage::Ready); let uuid = uuid_lock.read().unwrap().to_string(); send_done_once(sent_done_or_error, &uuid, channel.clone()).await; Ok(client_channel) diff --git a/src/rpc/dial_report.rs b/src/rpc/dial_report.rs new file mode 100644 index 0000000..7902e2c --- /dev/null +++ b/src/rpc/dial_report.rs @@ -0,0 +1,321 @@ +//! Best-effort reporting of WebRTC dial connection metadata to the Viam app signaling server. +//! +//! This is a port of goutils#583 (`rpc/wrtc_client_report.go`). After a WebRTC dial finishes — +//! success or failure — the client reports to the app signaling server it dialed through: the +//! furthest dial stage reached, the gRPC failure code, the dial duration, how the dial was +//! signaled, and the selected ICE candidate pair per side (host / stun / relay, plus the relay +//! address). Delivery is best-effort and runs in a detached background task so reporting never +//! adds latency to, or fails, a dial. +//! +//! Only the app signaling server implements `ReportConnectionMetadata`. A cloud-signaled dial +//! (routed through `app.viam.com` / `app.viam.dev`) reports over its own channel; a locally-signaled +//! dial reconstructs an authenticated connection to prod app, reusing the dial's credentials (Go's +//! `fixUpReportDialOpts`). rust-utils does not signal WebRTC over mDNS, so no report carries the +//! `MdnsLocal` path. See [`should_deliver`] for which outcomes are reported. + +use std::sync::atomic::{AtomicI32, Ordering}; +use std::time::Duration; + +use ::http::HeaderValue; +use ::webrtc::ice::candidate::{CandidatePairState, CandidateType}; +use ::webrtc::peer_connection::RTCPeerConnection; +use ::webrtc::stats::{StatsReport, StatsReportType}; +use tonic::transport::Channel; +use tower_http::auth::AddAuthorization; +use tower_http::set_header::SetRequestHeader; + +use crate::gen::proto::rpc::webrtc::v1::{ + signaling_service_client::SignalingServiceClient, ConnectionCandidate, ConnectionSignalingPath, + DialStage, IceCandidateType, ReportConnectionMetadataRequest, +}; + +/// The authenticated, rpc-host-stamped channel used to reach a signaling server, over which a +/// report can be delivered. +type SignalingChannel = AddAuthorization>; + +/// gRPC status code reported for a failed dial whose error is not a `tonic::Status`. Matches Go's +/// `status.Code` returning `codes.Unknown` for non-status errors. +const STATUS_CODE_UNKNOWN: i32 = 2; + +/// How long to wait for a report RPC before giving up. +const REPORT_TIMEOUT: Duration = Duration::from_secs(5); + +/// The Viam app signaling server hosts. A dial signaled through one of these is "cloud-signaled". +const VIAM_CLOUD_SIGNALING_HOSTS: [&str; 2] = ["app.viam.com", "app.viam.dev"]; + +/// Tracks the furthest dial checkpoint a WebRTC dial reached. It is advanced from several tasks +/// (the dial goroutine, the candidate-exchange task, and ICE/peer-connection callbacks), so it is +/// an atomic; `advance` only ever moves it forward. +pub(crate) struct StageTracker(AtomicI32); + +impl StageTracker { + pub(crate) fn new() -> Self { + StageTracker(AtomicI32::new(DialStage::Unspecified as i32)) + } + + /// Moves the reached stage forward to `stage` if `stage` is further than the current value. + pub(crate) fn advance(&self, stage: DialStage) { + let next = stage as i32; + let mut cur = self.0.load(Ordering::Acquire); + while next > cur { + match self + .0 + .compare_exchange_weak(cur, next, Ordering::AcqRel, Ordering::Acquire) + { + Ok(_) => break, + Err(actual) => cur = actual, + } + } + } + + /// The furthest stage reached so far, as the raw enum value. + pub(crate) fn reached(&self) -> i32 { + self.0.load(Ordering::Acquire) + } +} + +/// The env var that opts a process out of dial reporting (any non-empty value disables it). +const DISABLE_DIAL_REPORTING_ENV: &str = "VIAM_DISABLE_DIAL_REPORTING"; + +/// Whether a dial should deliver its connection report. +/// +/// Disabled under `cfg(test)` (the crate's own unit tests) and whenever `VIAM_DISABLE_DIAL_REPORTING` +/// is set to a non-empty value, so a detached report task can't outlive a test (mirrors Go +/// disabling reports in test binaries). `cfg(test)` does not cover integration tests in `tests/` or +/// downstream consumers, so test/CI harnesses that dial real robots should set the env var; +/// production consumers (release builds, no env var) report normally. +pub(crate) fn dial_reporting_enabled() -> bool { + if cfg!(test) { + return false; + } + let disabled = std::env::var(DISABLE_DIAL_REPORTING_ENV) + .map(|value| !value.is_empty()) + .unwrap_or(false); + !disabled +} + +/// Derives how a WebRTC dial was signaled from the signaling server host. rust-utils only signals +/// WebRTC over the cloud or a local signaling server (it does not yet support WebRTC over mDNS), +/// so this returns `CloudSignaled` or `Local` — never `MdnsLocal`. +pub(crate) fn classify_signaling_path(signaling_host: &str) -> ConnectionSignalingPath { + // Strip a trailing :port if present. (Signaling hosts are never bracketed IPv6, so a plain + // rsplit on ':' is safe.) + let host = signaling_host + .rsplit_once(':') + .map_or(signaling_host, |(host, _port)| host); + if VIAM_CLOUD_SIGNALING_HOSTS.contains(&host.to_ascii_lowercase().as_str()) { + ConnectionSignalingPath::CloudSignaled + } else { + ConnectionSignalingPath::Local + } +} + +/// Returns the (local, remote) candidate ids of the ICE candidate pair a WebRTC connection settled +/// on — the nominated pair in the succeeded state — or None if no such pair exists. +fn selected_candidate_pair(stats: &StatsReport) -> Option<(String, String)> { + stats.reports.values().find_map(|report| match report { + StatsReportType::CandidatePair(pair) + if pair.nominated && pair.state == CandidatePairState::Succeeded => + { + Some(( + pair.local_candidate_id.clone(), + pair.remote_candidate_id.clone(), + )) + } + _ => None, + }) +} + +/// Inspects the selected ICE candidate pair and classifies each side into a `ConnectionCandidate`. +/// Both are the default (type UNSPECIFIED) when `peer` is None (a failed dial) or no succeeded, +/// nominated pair exists. +pub(crate) async fn classify_connection( + peer: Option<&RTCPeerConnection>, +) -> (ConnectionCandidate, ConnectionCandidate) { + let Some(peer) = peer else { + return ( + ConnectionCandidate::default(), + ConnectionCandidate::default(), + ); + }; + let stats = peer.get_stats().await; + let Some((local_id, remote_id)) = selected_candidate_pair(&stats) else { + return ( + ConnectionCandidate::default(), + ConnectionCandidate::default(), + ); + }; + ( + classify_candidate(&stats, &local_id), + classify_candidate(&stats, &remote_id), + ) +} + +/// Maps a single ICE candidate stat to a `ConnectionCandidate`; a missing or unrecognized +/// candidate yields type UNSPECIFIED. Relay candidates carry the relay server address so the +/// signaling server can classify the relay provider. +fn classify_candidate(stats: &StatsReport, candidate_id: &str) -> ConnectionCandidate { + let candidate = match stats.reports.get(candidate_id) { + Some(StatsReportType::LocalCandidate(candidate)) + | Some(StatsReportType::RemoteCandidate(candidate)) => candidate, + _ => return ConnectionCandidate::default(), + }; + match candidate.candidate_type { + CandidateType::Host => ConnectionCandidate { + r#type: IceCandidateType::Host as i32, + relay_address: String::new(), + }, + CandidateType::ServerReflexive | CandidateType::PeerReflexive => ConnectionCandidate { + r#type: IceCandidateType::Stun as i32, + relay_address: String::new(), + }, + CandidateType::Relay => ConnectionCandidate { + r#type: IceCandidateType::Relay as i32, + relay_address: candidate.ip.clone(), + }, + CandidateType::Unspecified => ConnectionCandidate::default(), + } +} + +/// The gRPC status code to report for a failed dial: the `tonic::Status` code if the error chain +/// carries one, otherwise UNKNOWN (matching Go's `status.Code` for non-status errors). +pub(crate) fn failure_code(err: &anyhow::Error) -> i32 { + for cause in err.chain() { + if let Some(status) = cause.downcast_ref::() { + return status.code() as i32; + } + } + STATUS_CODE_UNKNOWN +} + +/// Whether a built report should actually be delivered. +/// +/// On a successful WebRTC dial only a READY report is truthful (a non-READY furthest stage on +/// success would count a spurious failure against a dial that succeeded). +/// +/// A failed WebRTC dial is reported only when the failure is terminal — i.e. there is no working +/// fallback. rust-utils falls back to a direct gRPC connection on any WebRTC failure, so: +/// - **cloud-signaled**: direct gRPC cannot reach a cloud robot, so the WebRTC failure fails the +/// dial and is reported; +/// - **local-signaled**: the direct gRPC connection to the robot generally succeeds, so the WebRTC +/// failure did not actually fail the dial — it is suppressed (mirrors Go suppressing a non-READY +/// report when the logical dial nonetheless succeeded). +pub(crate) fn should_deliver( + reached_stage: i32, + dial_succeeded: bool, + signaling_path: ConnectionSignalingPath, +) -> bool { + if dial_succeeded { + reached_stage == DialStage::Ready as i32 + } else { + signaling_path == ConnectionSignalingPath::CloudSignaled + } +} + +/// Delivers a single connection report over the (already authenticated, rpc-host-stamped) signaling +/// channel the dial used. Best-effort with a short timeout; failures are logged at debug and +/// otherwise swallowed. Intended to be spawned as a detached background task. +pub(crate) async fn send_dial_report( + channel: SignalingChannel, + request: ReportConnectionMetadataRequest, +) { + let reached_stage = request.reached_stage; + let mut client = SignalingServiceClient::new(channel); + match tokio::time::timeout(REPORT_TIMEOUT, client.report_connection_metadata(request)).await { + Ok(Ok(_)) => {} + Ok(Err(e)) => { + log::debug!("failed to report connection metadata (reached_stage={reached_stage}): {e}") + } + Err(_) => { + log::debug!("timed out reporting connection metadata (reached_stage={reached_stage})") + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn classify_signaling_path_cloud_vs_local() { + assert_eq!( + classify_signaling_path("app.viam.com:443"), + ConnectionSignalingPath::CloudSignaled + ); + assert_eq!( + classify_signaling_path("app.viam.com"), + ConnectionSignalingPath::CloudSignaled + ); + assert_eq!( + classify_signaling_path("APP.VIAM.DEV:443"), + ConnectionSignalingPath::CloudSignaled + ); + assert_eq!( + classify_signaling_path("app.viaminternal:8089"), + ConnectionSignalingPath::Local + ); + assert_eq!( + classify_signaling_path("localhost:8080"), + ConnectionSignalingPath::Local + ); + assert_eq!( + classify_signaling_path("10.1.2.3:443"), + ConnectionSignalingPath::Local + ); + } + + #[test] + fn should_deliver_suppresses_non_ready_on_success() { + use ConnectionSignalingPath::{CloudSignaled, Local}; + // Success: only a READY report is delivered, regardless of signaling path. + assert!(should_deliver(DialStage::Ready as i32, true, CloudSignaled)); + assert!(should_deliver(DialStage::Ready as i32, true, Local)); + assert!(!should_deliver( + DialStage::IceConnected as i32, + true, + CloudSignaled + )); + assert!(!should_deliver(DialStage::Unspecified as i32, true, Local)); + // Cloud failure: reported (direct gRPC can't reach a cloud robot, so it's terminal). + assert!(should_deliver( + DialStage::Unspecified as i32, + false, + CloudSignaled + )); + assert!(should_deliver( + DialStage::OfferSent as i32, + false, + CloudSignaled + )); + // Local failure: suppressed (the dial falls back to a working direct gRPC connection). + assert!(!should_deliver(DialStage::OfferSent as i32, false, Local)); + assert!(!should_deliver( + DialStage::IceConnected as i32, + false, + Local + )); + } + + #[test] + fn stage_tracker_only_moves_forward() { + let tracker = StageTracker::new(); + assert_eq!(tracker.reached(), DialStage::Unspecified as i32); + tracker.advance(DialStage::OfferSent); + assert_eq!(tracker.reached(), DialStage::OfferSent as i32); + // A lower stage does not regress the tracker. + tracker.advance(DialStage::SignalingConnected); + assert_eq!(tracker.reached(), DialStage::OfferSent as i32); + tracker.advance(DialStage::Ready); + assert_eq!(tracker.reached(), DialStage::Ready as i32); + } + + #[test] + fn failure_code_extracts_tonic_status() { + let status = tonic::Status::new(tonic::Code::PermissionDenied, "nope"); + let err = anyhow::Error::from(status).context("while dialing"); + assert_eq!(failure_code(&err), tonic::Code::PermissionDenied as i32); + + let plain = anyhow::anyhow!("some non-status error"); + assert_eq!(failure_code(&plain), STATUS_CODE_UNKNOWN); + } +} diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index 7a0ac1d..a83252d 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -3,5 +3,6 @@ mod base_stream; mod client_channel; mod client_stream; pub mod dial; +mod dial_report; pub mod log_prefixes; mod webrtc; diff --git a/src/rpc/webrtc.rs b/src/rpc/webrtc.rs index 5e9733f..ca4c217 100644 --- a/src/rpc/webrtc.rs +++ b/src/rpc/webrtc.rs @@ -1,5 +1,6 @@ +use super::dial_report::StageTracker; use super::log_prefixes; -use crate::gen::proto::rpc::webrtc::v1::{IceServer, ResponseTrailers, WebRtcConfig}; +use crate::gen::proto::rpc::webrtc::v1::{DialStage, IceServer, ResponseTrailers, WebRtcConfig}; use anyhow::Result; use bytes::Bytes; use core::fmt; @@ -276,6 +277,7 @@ fn create_invalid_sdp_err(err: serde_json::error::Error) -> webrtc::Error { pub(crate) async fn new_peer_connection_for_client( config: RTCConfiguration, disable_trickle_ice: bool, + stage_tracker: Arc, ) -> Result<(Arc, Arc)> { let web_api = new_webrtc_api()?; let peer_connection = Arc::new(web_api.new_peer_connection(config).await?); @@ -294,9 +296,13 @@ pub(crate) async fn new_peer_connection_for_client( peer_connection.on_peer_connection_state_change(Box::new( move |connection: RTCPeerConnectionState| { + let stage_tracker = stage_tracker.clone(); log::info!("peer connection state change: {connection}"); if connection == RTCPeerConnectionState::Connected { log::debug!("{}", log_prefixes::DIALED_WEBRTC); + // The peer connection reaching Connected means ICE + DTLS completed; the data + // channel is not necessarily open yet (that is READY). + stage_tracker.advance(DialStage::DtlsConnected); } Box::pin(async move {}) }, diff --git a/tests/echo_test.rs b/tests/echo_test.rs index 3109c40..94c385c 100644 --- a/tests/echo_test.rs +++ b/tests/echo_test.rs @@ -9,7 +9,14 @@ use viam_rust_utils::gen::proto::rpc::examples::echo::v1::{ }; use viam_rust_utils::rpc::dial; +// Keep the best-effort dial connection reports (see rpc::dial_report) from firing during tests, so +// no detached report task outlives a test. +fn disable_dial_reporting() { + env::set_var("VIAM_DISABLE_DIAL_REPORTING", "1"); +} + async fn dial_direct() -> Result { + disable_dial_reporting(); let port = env::var("SERVER_PORT").unwrap().to_owned(); let uri = ["localhost:".to_string(), port].join(""); @@ -107,6 +114,7 @@ async fn test_dial_direct_bidi() -> Result<()> { } async fn dial_webrtc() -> Result { + disable_dial_reporting(); let port = env::var("SERVER_PORT").unwrap().to_owned(); let uri = ["localhost:".to_string(), port].join("");