viam_rust_utils/rpc/
dial.rs

1use super::{
2    client_channel::*,
3    log_prefixes,
4    webrtc::{webrtc_action_with_timeout, Options},
5};
6use crate::gen::google;
7use crate::gen::proto::rpc::v1::{
8    auth_service_client::AuthServiceClient, AuthenticateRequest, Credentials,
9};
10use crate::gen::proto::rpc::webrtc::v1::{
11    call_response::Stage, call_update_request::Update,
12    signaling_service_client::SignalingServiceClient, CallUpdateRequest,
13    OptionalWebRtcConfigRequest, OptionalWebRtcConfigResponse,
14};
15use crate::gen::proto::rpc::webrtc::v1::{
16    CallRequest, IceCandidate, Metadata, RequestHeaders, Strings,
17};
18use crate::rpc::webrtc;
19use ::http::header::HeaderName;
20use ::http::{
21    uri::{Authority, Parts, PathAndQuery, Scheme},
22    HeaderValue, Version,
23};
24use ::viam_mdns::{discover, RecordKind, Response};
25use ::webrtc::ice_transport::{
26    ice_candidate::{RTCIceCandidate, RTCIceCandidateInit},
27    ice_connection_state::RTCIceConnectionState,
28};
29use ::webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
30use anyhow::{Context, Result};
31use core::fmt;
32use futures::stream::FuturesUnordered;
33use futures_util::{pin_mut, stream::StreamExt};
34use local_ip_address::list_afinet_netifas;
35use std::{
36    collections::HashMap,
37    net::{IpAddr, Ipv4Addr},
38    sync::{
39        atomic::{AtomicBool, Ordering},
40        Arc, Mutex, RwLock,
41    },
42    task::{Context as TaskContext, Poll},
43    time::{Duration, Instant},
44};
45use tokio::sync::{mpsc, watch};
46use tonic::body::BoxBody;
47use tonic::codegen::BoxFuture;
48use tonic::transport::{Body, Channel, ClientTlsConfig, Uri};
49
50use tower::{Service, ServiceBuilder};
51use tower_http::auth::AddAuthorization;
52use tower_http::auth::AddAuthorizationLayer;
53use tower_http::set_header::{SetRequestHeader, SetRequestHeaderLayer};
54
55// gRPC status codes
56const STATUS_CODE_OK: i32 = 0;
57const STATUS_CODE_UNKNOWN: i32 = 2;
58const STATUS_CODE_RESOURCE_EXHAUSTED: i32 = 8;
59
60pub const VIAM_MDNS_SERVICE_NAME: &'static str = "_rpc._tcp.local";
61
62type SecretType = String;
63
64#[derive(Clone)]
65/// A communication channel to a given uri. The channel is either a direct tonic channel,
66/// or a webRTC channel.
67pub enum ViamChannel {
68    Direct(Channel),
69    DirectPreAuthorized(AddAuthorization<SetRequestHeader<Channel, HeaderValue>>),
70    WebRTC(Arc<WebRTCClientChannel>),
71}
72
73#[derive(Debug, Clone)]
74pub struct RPCCredentials {
75    entity: Option<String>,
76    credentials: Credentials,
77}
78
79impl RPCCredentials {
80    pub fn new(entity: Option<String>, r#type: SecretType, payload: String) -> Self {
81        Self {
82            credentials: Credentials { r#type, payload },
83            entity,
84        }
85    }
86}
87
88impl ViamChannel {
89    async fn create_resp(
90        channel: &mut Arc<WebRTCClientChannel>,
91        stream: crate::gen::proto::rpc::webrtc::v1::Stream,
92        request: http::Request<BoxBody>,
93        response: http::response::Builder,
94    ) -> http::Response<Body> {
95        let (parts, body) = request.into_parts();
96        let mut status_code = STATUS_CODE_OK;
97        let stream_id = stream.id;
98        let metadata = Some(metadata_from_parts(&parts));
99        let headers = RequestHeaders {
100            method: parts
101                .uri
102                .path_and_query()
103                .map(PathAndQuery::to_string)
104                .unwrap_or_default(),
105            metadata,
106            timeout: None,
107        };
108
109        if let Err(e) = channel.write_headers(&stream, headers).await {
110            log::error!("error writing headers: {e}");
111            channel.close_stream_with_recv_error(stream_id, e);
112            status_code = STATUS_CODE_UNKNOWN;
113        }
114
115        // Pump the request body to the data channel on its own task so that response
116        // frames (delivered async via on_channel_message into the resp body) can flow
117        // back to the caller WHILE requests are still being sent. Required for
118        // client-streaming and bidi; previously we drained the whole body with
119        // `to_bytes` before returning the response, which serialized send-then-receive.
120        if status_code == STATUS_CODE_OK {
121            let send_channel = channel.clone();
122            let send_stream = stream.clone();
123            tokio::spawn(async move {
124                if let Err(e) = pump_request_body(&send_channel, &send_stream, body).await {
125                    log::error!("error sending message: {e}");
126                    send_channel.close_stream_with_recv_error(stream_id, e);
127                }
128            });
129        }
130
131        let body = match channel.resp_body_from_stream(stream_id) {
132            Ok(body) => body,
133            Err(e) => {
134                log::error!("error receiving response from stream: {e}");
135                channel.close_stream_with_recv_error(stream_id, e);
136                status_code = STATUS_CODE_UNKNOWN;
137                Body::empty()
138            }
139        };
140
141        let response = if status_code != STATUS_CODE_OK {
142            response.header("grpc-status", &status_code.to_string())
143        } else {
144            response
145        };
146
147        response.body(body).unwrap()
148    }
149}
150
151/// Reads `body` frame-by-frame, reassembling complete gRPC-framed messages across
152/// hyper frame boundaries and forwarding each to the data channel as soon as it is
153/// complete, then signals end of stream. Streaming the body (rather than buffering it
154/// whole with `to_bytes`) is what lets client requests interleave with server
155/// responses, enabling client-streaming and bidi.
156async fn pump_request_body(
157    channel: &Arc<WebRTCClientChannel>,
158    stream: &crate::gen::proto::rpc::webrtc::v1::Stream,
159    mut body: BoxBody,
160) -> Result<()> {
161    use http_body::Body as _;
162
163    // Abort the pump if the underlying connection dies. `body.data()` polls the
164    // app-side request body, which is independent of the data channel, so without
165    // this a dead connection with an idle-but-open send half would park this task
166    // forever while it pins the channel Arc alive. Built once outside the loop so the
167    // hot send path does not re-register a waiter on every message.
168    let cancelled = channel.base_channel.closed_token.clone().cancelled_owned();
169    tokio::pin!(cancelled);
170
171    let mut buf: Vec<u8> = Vec::new();
172    loop {
173        let chunk = tokio::select! {
174            biased;
175            _ = &mut cancelled => {
176                return Err(anyhow::anyhow!(
177                    "connection closed before request body completed; aborting send"
178                ));
179            }
180            chunk = body.data() => chunk,
181        };
182        let chunk = match chunk {
183            Some(chunk) => chunk.map_err(|e| anyhow::anyhow!("error reading request body: {e}"))?,
184            None => break, // body complete
185        };
186        buf.extend_from_slice(&chunk);
187
188        // Emit every complete gRPC message currently buffered. A gRPC frame is a
189        // 1-byte compression flag + 4-byte big-endian length + payload.
190        loop {
191            if buf.len() < 5 {
192                break;
193            }
194            let len = u32::from_be_bytes(buf[1..5].try_into().unwrap()) as usize;
195            let total = 5 + len;
196            if buf.len() < total {
197                break; // message spans further frames; wait for the rest
198            }
199            let msg: Vec<u8> = buf.drain(..total).collect();
200            channel.write_grpc_message(stream, &msg).await?;
201        }
202    }
203
204    if !buf.is_empty() {
205        return Err(anyhow::anyhow!(
206            "request body ended with {} trailing bytes of an incomplete message",
207            buf.len()
208        ));
209    }
210
211    channel.write_eos(stream).await
212}
213
214impl Service<http::Request<BoxBody>> for ViamChannel {
215    type Response = http::Response<Body>;
216    type Error = tonic::transport::Error;
217    type Future = BoxFuture<Self::Response, Self::Error>;
218
219    fn poll_ready(&mut self, cx: &mut TaskContext<'_>) -> Poll<Result<(), Self::Error>> {
220        match self {
221            Self::Direct(channel) => channel.poll_ready(cx),
222            Self::DirectPreAuthorized(channel) => channel.poll_ready(cx),
223            Self::WebRTC(_channel) => Poll::Ready(Ok(())),
224        }
225    }
226
227    fn call(&mut self, request: http::Request<BoxBody>) -> Self::Future {
228        match self {
229            Self::Direct(channel) => Box::pin(channel.call(request)),
230            Self::DirectPreAuthorized(channel) => Box::pin(channel.call(request)),
231            Self::WebRTC(channel) => {
232                let mut channel = channel.clone();
233                let fut = async move {
234                    let response = http::response::Response::builder()
235                        // standardized gRPC headers.
236                        .header("content-type", "application/grpc")
237                        .version(Version::HTTP_2);
238
239                    match channel.new_stream() {
240                        Err(e) => {
241                            log::error!("{e}");
242                            let response = response
243                                .header("grpc-status", &STATUS_CODE_RESOURCE_EXHAUSTED.to_string())
244                                .body(Body::default())
245                                .unwrap();
246
247                            Ok(response)
248                        }
249                        Ok(stream) => {
250                            Ok(Self::create_resp(&mut channel, stream, request, response).await)
251                        }
252                    }
253                };
254                Box::pin(fut)
255            }
256        }
257    }
258}
259
260/// Options for modifying the connection parameters
261#[derive(Debug)]
262pub struct DialOptions {
263    credentials: Option<RPCCredentials>,
264    webrtc_options: Option<Options>,
265    uri: Option<Parts>,
266    disable_mdns: bool,
267    allow_downgrade: bool,
268    insecure: bool,
269    signaling_server_override: Option<String>,
270}
271#[derive(Clone)]
272pub struct WantsCredentials(());
273#[derive(Clone)]
274pub struct WantsUri(());
275#[derive(Clone)]
276pub struct WithCredentials(());
277#[derive(Clone)]
278pub struct WithoutCredentials(());
279
280pub trait AuthMethod {}
281impl AuthMethod for WithCredentials {}
282impl AuthMethod for WithoutCredentials {}
283/// A DialBuilder allows us to set options before establishing a connection to a server
284#[allow(dead_code)]
285pub struct DialBuilder<T> {
286    state: T,
287    config: DialOptions,
288}
289
290impl<T> fmt::Debug for DialBuilder<T> {
291    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
292        f.debug_struct("Dial")
293            .field("State", &format_args!("{}", &std::any::type_name::<T>()))
294            .field("Opt", &format_args!("{:?}", self.config))
295            .finish()
296    }
297}
298
299impl DialOptions {
300    /// Creates a new DialBuilder
301    pub fn builder() -> DialBuilder<WantsUri> {
302        DialBuilder {
303            state: WantsUri(()),
304            config: DialOptions {
305                credentials: None,
306                uri: None,
307                allow_downgrade: false,
308                disable_mdns: false,
309                insecure: false,
310                webrtc_options: None,
311                signaling_server_override: None,
312            },
313        }
314    }
315}
316
317impl DialBuilder<WantsUri> {
318    /// Sets the uri to connect to
319    pub fn uri(self, uri: &str) -> DialBuilder<WantsCredentials> {
320        let uri_parts = uri_parts_with_defaults(uri);
321        DialBuilder {
322            state: WantsCredentials(()),
323            config: DialOptions {
324                credentials: None,
325                uri: Some(uri_parts),
326                allow_downgrade: false,
327                disable_mdns: false,
328                insecure: false,
329                webrtc_options: None,
330                signaling_server_override: None,
331            },
332        }
333    }
334}
335impl DialBuilder<WantsCredentials> {
336    /// Tells connecting logic to not expect/require credentials
337    pub fn without_credentials(self) -> DialBuilder<WithoutCredentials> {
338        DialBuilder {
339            state: WithoutCredentials(()),
340            config: DialOptions {
341                credentials: None,
342                uri: self.config.uri,
343                allow_downgrade: false,
344                disable_mdns: false,
345                insecure: false,
346                webrtc_options: None,
347                signaling_server_override: None,
348            },
349        }
350    }
351    /// Sets credentials to use when connecting
352    pub fn with_credentials(self, creds: RPCCredentials) -> DialBuilder<WithCredentials> {
353        DialBuilder {
354            state: WithCredentials(()),
355            config: DialOptions {
356                credentials: Some(creds),
357                uri: self.config.uri,
358                allow_downgrade: false,
359                disable_mdns: false,
360                insecure: false,
361                webrtc_options: None,
362                signaling_server_override: None,
363            },
364        }
365    }
366}
367
368impl<T: AuthMethod> DialBuilder<T> {
369    /// Attempts to connect insecurely with scheme of HTTP as a default
370    pub fn insecure(mut self) -> Self {
371        self.config.insecure = true;
372        self
373    }
374    /// Allows for downgrading and attempting to connect via HTTP if HTTPS fails
375    pub fn allow_downgrade(mut self) -> Self {
376        self.config.allow_downgrade = true;
377        self
378    }
379    /// Disables connection via mDNS
380    pub fn disable_mdns(mut self) -> Self {
381        self.config.disable_mdns = true;
382        self
383    }
384
385    /// Overrides any default connection behavior, forcing direct connection. Note that
386    /// the connection itself will fail if it is between a client and server on separate
387    /// networks and not over webRTC
388    pub fn disable_webrtc(mut self) -> Self {
389        let webrtc_options = Options::default().disable_webrtc();
390        self.config.webrtc_options = Some(webrtc_options);
391        self
392    }
393
394    /// Forces ICE transport policy to relay-only so only TURN candidates are used.
395    /// Useful for testing relay connectivity through a TURN server.
396    pub fn force_relay(mut self) -> Self {
397        self.config
398            .webrtc_options
399            .get_or_insert_with(Options::default)
400            .force_relay = true;
401        self
402    }
403
404    /// Strips TURN servers from the ICE config so only host and server-reflexive
405    /// candidates are used. Useful for testing direct connectivity without relay fallback.
406    pub fn force_p2p(mut self) -> Self {
407        self.config
408            .webrtc_options
409            .get_or_insert_with(Options::default)
410            .force_p2p = true;
411        self
412    }
413
414    /// Filters the signaling server's TURN list to only the server whose parsed URI
415    /// matches (compared by scheme, host, port, and transport — defaulting transport
416    /// to UDP if unspecified). Example: "turn:turn.viam.com:443"
417    pub fn turn_uri(mut self, uri: String) -> Self {
418        self.config
419            .webrtc_options
420            .get_or_insert_with(Options::default)
421            .turn_uri = Some(uri);
422        self
423    }
424
425    /// Overrides the signaling server address used for WebRTC negotiation.
426    pub fn signaling_server(mut self, address: String) -> Self {
427        self.config.signaling_server_override = Some(address);
428        self
429    }
430
431    async fn get_addr_from_interface(
432        iface: (&str, Vec<&IpAddr>),
433        candidates: &Vec<String>,
434        local_ipv4s: &std::collections::HashSet<Ipv4Addr>,
435    ) -> Option<String> {
436        let addresses: Vec<Ipv4Addr> = iface
437            .1
438            .iter()
439            .filter_map(|ip| match ip {
440                IpAddr::V4(v4) => Some(*v4),
441                IpAddr::V6(_) => None,
442            })
443            .collect();
444
445        let mut resp: Option<Response> = None;
446        for ipv4 in addresses {
447            for candidate in candidates {
448                let discovery = match discover::interface_with_loopback(
449                    VIAM_MDNS_SERVICE_NAME,
450                    Duration::from_millis(250),
451                    ipv4,
452                ) {
453                    Ok(d) => d,
454                    Err(e) => {
455                        log::debug!("mDNS socket error on {ipv4}: {e}");
456                        continue;
457                    }
458                };
459                let stream = discovery.listen();
460                pin_mut!(stream);
461                while let Some(Ok(response)) = stream.next().await {
462                    if let Some(hostname) = response.hostname() {
463                        // Machine uris come in local ("my-cool-robot.abcdefg.local.viam.cloud")
464                        // and non-local ("my-cool-robot.abcdefg.viam.cloud") forms. Sometimes
465                        // (namely with micro-rdk), our mdns query can only see one (the local) version.
466                        // However, users are typically passing the non-local version. By splitting at
467                        // "viam" and taking the only the first value, we can still search for
468                        // candidates based on the actual "my-cool-robot" name without being opinionated
469                        // on whether the candidate is locally named or not.
470                        let local_agnostic_candidate = candidate.as_str().split("viam").next()?;
471                        log::debug!(
472                            "mDNS response on {ipv4}: hostname={hostname:?}, candidate={candidate:?}, local_agnostic={local_agnostic_candidate:?}, matches={}",
473                            hostname.contains(local_agnostic_candidate)
474                        );
475                        if hostname.contains(local_agnostic_candidate) {
476                            resp = Some(response);
477                            break;
478                        }
479                    } else {
480                        log::debug!(
481                            "mDNS response on {ipv4}: no hostname (no PTR record); answers={:?}",
482                            response.answers
483                        );
484                    }
485                    if resp.is_some() {
486                        break;
487                    }
488                }
489            }
490        }
491
492        let resp = resp?;
493        let mut has_grpc = false;
494        let mut has_webrtc = false;
495        for field in resp.txt_records() {
496            has_grpc = has_grpc || field.contains("grpc");
497            has_webrtc = has_webrtc || field.contains("webrtc");
498        }
499
500        // Log all records in the response for diagnostics.
501        log::debug!(
502            "mDNS matched response records: {:?}",
503            resp.records().collect::<Vec<_>>()
504        );
505
506        // Select the best IP from the mDNS response using a three-tier preference:
507        //
508        // 1. Non-loopback IP that is currently assigned to one of our own network
509        //    interfaces.  This handles the same-machine case (client and robot on
510        //    the same host) while avoiding stale IPs that were valid when
511        //    viam-server started but are now unreachable (e.g. a WiFi address after
512        //    WiFi was disconnected).
513        //
514        // 2. Any IP (including loopback) that is currently assigned to one of our
515        //    interfaces.  This catches 127.0.0.1 when offline on the same machine:
516        //    127.0.0.1 is excluded from tier 1 by the !is_loopback() guard, but it
517        //    is always in local_ipv4s, so it is correctly preferred here over a
518        //    stale non-loopback address that is no longer assigned.
519        //
520        // 3. Last resort: any advertised IPv4, for the common case of connecting to
521        //    a robot on a separate machine (its IP will never appear in local_ipv4s).
522        let ip_addr = resp
523            .records()
524            .filter_map(|r| match r.kind {
525                RecordKind::A(addr) if !addr.is_loopback() && local_ipv4s.contains(&addr) => {
526                    Some(addr)
527                }
528                _ => None,
529            })
530            .next()
531            .or_else(|| {
532                resp.records()
533                    .find_map(|r| match r.kind {
534                        RecordKind::A(addr) if local_ipv4s.contains(&addr) => Some(addr),
535                        _ => None,
536                    })
537                    .or_else(|| {
538                        resp.records().find_map(|r| match r.kind {
539                            RecordKind::A(addr) => Some(addr),
540                            _ => None,
541                        })
542                    })
543            });
544
545        if !(has_grpc || has_webrtc) || ip_addr.is_none() {
546            return None;
547        }
548        let mut local_addr = ip_addr?.to_string();
549        local_addr.push(':');
550        local_addr.push_str(&resp.port()?.to_string());
551        log::debug!("mDNS resolved address: {local_addr}");
552        Some(local_addr)
553    }
554
555    fn duplicate_uri(&self) -> Option<Parts> {
556        match &self.config.uri {
557            None => None,
558            Some(uri) => duplicate_uri(uri),
559        }
560    }
561
562    async fn get_mdns_uri(&self) -> Option<Parts> {
563        log::debug!("{}", log_prefixes::MDNS_QUERY_ATTEMPT);
564        if self.config.disable_mdns {
565            return None;
566        }
567
568        let mut uri = self.duplicate_uri()?;
569        let candidate = uri.authority.clone()?.to_string();
570
571        let candidates: Vec<String> = vec![candidate.replace('.', "-"), candidate];
572
573        let ifaces = list_afinet_netifas().ok()?;
574
575        // Collect all local IPv4 addresses for use in get_addr_from_interface, which prefers
576        // mDNS response IPs that are currently assigned to one of our own interfaces.
577        // viam-server may advertise a stale IP (e.g. a WiFi address from when it started,
578        // now unreachable because WiFi is off); filtering by current interface addresses
579        // avoids connecting to an unreachable address.
580        let local_ipv4s: std::collections::HashSet<Ipv4Addr> = ifaces
581            .iter()
582            .filter_map(|(_, ip)| match ip {
583                IpAddr::V4(v4) => Some(*v4),
584                _ => None,
585            })
586            .collect();
587
588        let ifaces: HashMap<&str, Vec<&IpAddr>> =
589            ifaces.iter().fold(HashMap::new(), |mut map, (k, v)| {
590                map.entry(k).or_default().push(v);
591                map
592            });
593
594        let mut iface_futures = FuturesUnordered::new();
595        for iface in ifaces {
596            iface_futures.push(Self::get_addr_from_interface(
597                iface,
598                &candidates,
599                &local_ipv4s,
600            ));
601        }
602
603        let mut local_addr: Option<String> = None;
604        while let Some(maybe_addr) = iface_futures.next().await {
605            if maybe_addr.is_some() {
606                local_addr = maybe_addr;
607                break;
608            }
609        }
610        let local_addr = match local_addr {
611            None => {
612                log::debug!("Unable to connect via mDNS");
613                return None;
614            }
615            Some(addr) => {
616                log::debug!("{}: {addr}", log_prefixes::MDNS_ADDRESS_FOUND);
617                addr
618            }
619        };
620
621        let auth = local_addr.parse::<Authority>().ok()?;
622        uri.authority = Some(auth);
623
624        Some(uri)
625    }
626
627    async fn create_channel(
628        allow_downgrade: bool,
629        domain: &str,
630        uri: Uri,
631        for_mdns: bool,
632    ) -> Result<Channel> {
633        if for_mdns {
634            let host = uri.host().unwrap_or("");
635            // viam-server serves TLS gRPC on the port it advertises via mDNS, including on
636            // loopback.  `domain` is the robot's canonical hostname (e.g.
637            // `my-robot.abcdefg.viam.cloud`) used for SNI and SAN verification.
638            log::debug!("mDNS create_channel: connecting to {host} with TLS");
639            let tls_config = ClientTlsConfig::new().domain_name(domain);
640            let mut parts = uri.clone().into_parts();
641            parts.scheme = Some(Scheme::HTTPS);
642            let tls_uri = Uri::from_parts(parts)?;
643            match Channel::builder(tls_uri.clone())
644                .tls_config(tls_config)?
645                .connect()
646                .await
647                .with_context(|| format!("Connecting to {:?}", tls_uri))
648            {
649                Ok(channel) => return Ok(channel),
650                // When the caller allows insecure/downgrade, retry the local connection over plaintext
651                // instead of failing. A local viam-server with no TLS serves plain gRPC on the advertised
652                // port, so the TLS handshake above fails against it.
653                Err(e) => {
654                    if allow_downgrade {
655                        let mut parts = uri.into_parts();
656                        parts.scheme = Some(Scheme::HTTP);
657                        let uri = Uri::from_parts(parts)?;
658                        log::debug!(
659                            "mDNS TLS connect failed ({e:#}); downgrading to plaintext h2c {uri:?}"
660                        );
661                        return Channel::builder(uri.clone())
662                            .connect()
663                            .await
664                            .with_context(|| format!("Connecting to {:?}", uri));
665                    }
666                    return Err(e);
667                }
668            }
669        }
670
671        let chan = match Channel::builder(uri.clone())
672            .connect()
673            .await
674            .with_context(|| format!("Connecting to {:?}", uri.clone()))
675        {
676            Ok(c) => c,
677            Err(e) => {
678                if allow_downgrade {
679                    let mut uri_parts = uri.clone().into_parts();
680                    uri_parts.scheme = Some(Scheme::HTTP);
681                    let uri = Uri::from_parts(uri_parts)?;
682                    Channel::builder(uri).connect().await?
683                } else {
684                    return Err(anyhow::anyhow!(e));
685                }
686            }
687        };
688        Ok(chan)
689    }
690}
691
692impl DialBuilder<WithoutCredentials> {
693    fn clone(&self) -> Self {
694        DialBuilder {
695            state: WithoutCredentials(()),
696            config: DialOptions {
697                credentials: None,
698                webrtc_options: self.config.webrtc_options.clone(),
699                uri: self.duplicate_uri(),
700                disable_mdns: self.config.disable_mdns,
701                allow_downgrade: self.config.allow_downgrade,
702                insecure: self.config.insecure,
703                signaling_server_override: self.config.signaling_server_override.clone(),
704            },
705        }
706    }
707
708    /// attempts to establish a connection without credentials to the DialBuilder's given uri
709    async fn connect_inner(
710        self,
711        mdns_uri: Option<Parts>,
712        mut original_uri_parts: Parts,
713    ) -> Result<ViamChannel> {
714        let webrtc_options = self.config.webrtc_options;
715        let disable_webrtc = match &webrtc_options {
716            Some(options) => options.disable_webrtc,
717            None => false,
718        };
719        if self.config.insecure {
720            original_uri_parts.scheme = Some(Scheme::HTTP);
721        }
722        let original_uri = Uri::from_parts(original_uri_parts)?;
723        let uri2 = original_uri.clone();
724        let uri = infer_remote_uri_from_authority(
725            original_uri,
726            self.config.signaling_server_override.as_deref(),
727        );
728        let domain = uri2.authority().to_owned().unwrap().as_str();
729
730        let mdns_uri = mdns_uri.and_then(|p| Uri::from_parts(p).ok());
731        let attempting_mdns = mdns_uri.is_some();
732        if attempting_mdns {
733            log::debug!("Attempting to connect via mDNS");
734        } else {
735            log::debug!("Attempting to connect");
736        }
737
738        let channel = match mdns_uri {
739            Some(uri) => Self::create_channel(self.config.allow_downgrade, domain, uri, true).await,
740            // not actually an error necessarily, but we want to ensure that a channel is still
741            // created with the default uri
742            None => Err(anyhow::anyhow!("")),
743        };
744
745        let channel = match channel {
746            Ok(c) => {
747                log::debug!("Connected via mDNS");
748                c
749            }
750            Err(e) => {
751                if attempting_mdns {
752                    // mDNS found the robot but the connection failed — don't fall through to
753                    // the remote/signaling URI here.  The parallel `without_mdns` branch in
754                    // `connect()` is already handling that case.  Returning an error lets the
755                    // select! loop surface whichever branch succeeds, and avoids a spurious
756                    // connection attempt to app.viam.com when the device is offline.
757                    log::debug!("Unable to connect via mDNS. Error: {e:#}");
758                    return Err(e);
759                }
760                Self::create_channel(self.config.allow_downgrade, domain, uri.clone(), false)
761                    .await?
762            }
763        };
764
765        // TODO (RSDK-517) make maybe_connect_via_webrtc take a more generic type so we don't
766        // need to add these dummy layers.
767        let intercepted_channel = ServiceBuilder::new()
768            .layer(AddAuthorizationLayer::basic(
769                "fake username",
770                "fake password",
771            ))
772            .layer(SetRequestHeaderLayer::overriding(
773                HeaderName::from_static("rpc-host"),
774                HeaderValue::from_str(domain)?,
775            ))
776            .service(channel.clone());
777
778        // TODO (RSDK-14026): support WebRTC over mDNS for offline connections (e.g. video streaming).
779        if disable_webrtc || attempting_mdns {
780            log::debug!("{}", log_prefixes::DIALED_GRPC);
781            Ok(ViamChannel::Direct(channel.clone()))
782        } else {
783            match maybe_connect_via_webrtc(uri, intercepted_channel.clone(), webrtc_options).await {
784                Ok(webrtc_channel) => Ok(ViamChannel::WebRTC(webrtc_channel)),
785                Err(e) => {
786                    log::error!("error connecting via webrtc: {e}. Attempting to connect directly");
787                    log::debug!("{}", log_prefixes::DIALED_GRPC);
788                    Ok(ViamChannel::Direct(channel.clone()))
789                }
790            }
791        }
792    }
793
794    async fn connect_mdns(self, original_uri: Parts) -> Result<ViamChannel> {
795        let mdns_uri =
796            webrtc::action_with_timeout(self.get_mdns_uri(), Duration::from_millis(1500))
797                .await
798                .ok()
799                .flatten()
800                .ok_or(anyhow::anyhow!(
801                    "Unable to establish connection via mDNS; uri not found"
802                ))?;
803
804        self.connect_inner(Some(mdns_uri), original_uri).await
805    }
806
807    pub async fn connect(self) -> Result<ViamChannel> {
808        log::debug!("{}", log_prefixes::DIAL_ATTEMPT);
809        let original_uri = self.duplicate_uri().ok_or(anyhow::anyhow!(
810            "Attempting to connect but there was no uri"
811        ))?;
812        let original_uri2 = duplicate_uri(&original_uri).ok_or(anyhow::anyhow!(
813            "Attempting to connect but there was no uri"
814        ))?;
815
816        let skip_mdns = self.config.disable_mdns;
817
818        // We want to short circuit and return the first `Ok` result from our connection
819        // attempts, which `tokio::select!` does great. Buuuuut, we don't want to
820        // abandon the `Err` results, and we want to provide comprehensive logging for
821        // debugging purposes. Hence the loop and pinning. The pinning lets us reference
822        // the same future multiple times, while the loop lets us immediately return on the
823        // first `Ok` result while still seeing and logging any error results.
824        //
825        // When mDNS is skipped (disable_mdns), with_mdns_err is pre-set so
826        // the select guard disables that branch and only the direct connection is attempted.
827        tokio::pin! {
828            let with_mdns = self.clone().connect_mdns(original_uri);
829            let without_mdns = self.connect_inner(None, original_uri2);
830        }
831        let mut with_mdns_err: Option<anyhow::Error> =
832            skip_mdns.then(|| anyhow::anyhow!("mDNS skipped"));
833        let mut without_mdns_err: Option<anyhow::Error> = None;
834        while with_mdns_err.is_none() || without_mdns_err.is_none() {
835            tokio::select! {
836                with_mdns = &mut with_mdns, if with_mdns_err.is_none() => {
837                    match with_mdns {
838                        Ok(chan) => return Ok(chan),
839                        Err(e) => {
840                            log::debug!("Error connecting with mdns: {e}");
841                            with_mdns_err = Some(e);
842                        }
843                    }
844                }
845                without_mdns = &mut without_mdns, if without_mdns_err.is_none() => {
846                    match without_mdns {
847                        Ok(chan) => return Ok(chan),
848                        Err(e) => {
849                            log::debug!("Error connecting without mdns: {e}");
850                            without_mdns_err = Some(e);
851                        }
852                    }
853                }
854            }
855        }
856        Err(anyhow::anyhow!(
857            "Unable to connect with or without mdns.
858                    with_mdns err: {with_mdns_err:?}
859                    without_mdns err: {without_mdns_err:?}"
860        ))
861    }
862}
863
864async fn get_auth_token(
865    channel: &mut Channel,
866    creds: Credentials,
867    entity: String,
868) -> Result<String> {
869    let mut auth_service = AuthServiceClient::new(channel);
870    let req = AuthenticateRequest {
871        entity,
872        credentials: Some(creds),
873    };
874
875    let rsp = auth_service.authenticate(req).await?;
876    Ok(rsp.into_inner().access_token)
877}
878
879impl DialBuilder<WithCredentials> {
880    fn clone(&self) -> Self {
881        DialBuilder {
882            state: WithCredentials(()),
883            config: DialOptions {
884                credentials: self.config.credentials.clone(),
885                webrtc_options: self.config.webrtc_options.clone(),
886                uri: self.duplicate_uri(),
887                disable_mdns: self.config.disable_mdns,
888                allow_downgrade: self.config.allow_downgrade,
889                insecure: self.config.insecure,
890                signaling_server_override: self.config.signaling_server_override.clone(),
891            },
892        }
893    }
894
895    async fn connect_inner(
896        self,
897        mdns_uri: Option<Parts>,
898        mut original_uri_parts: Parts,
899    ) -> Result<ViamChannel> {
900        let is_insecure = self.config.insecure;
901
902        let webrtc_options = self.config.webrtc_options;
903        let disable_webrtc = match &webrtc_options {
904            Some(options) => options.disable_webrtc,
905            None => false,
906        };
907
908        if is_insecure {
909            original_uri_parts.scheme = Some(Scheme::HTTP);
910        }
911
912        let original_uri = Uri::from_parts(original_uri_parts)?;
913
914        let domain = original_uri.authority().unwrap().to_string();
915        let uri_for_auth = infer_remote_uri_from_authority(
916            original_uri.clone(),
917            self.config.signaling_server_override.as_deref(),
918        );
919
920        let mdns_uri = mdns_uri.and_then(|p| Uri::from_parts(p).ok());
921        let attempting_mdns = mdns_uri.is_some();
922
923        let allow_downgrade = self.config.allow_downgrade;
924        if attempting_mdns {
925            log::debug!("Attempting to connect via mDNS");
926        } else {
927            log::debug!("Attempting to connect");
928        }
929        let channel = match mdns_uri {
930            Some(uri) => Self::create_channel(allow_downgrade, &domain, uri, true).await,
931            // not actually an error necessarily, but we want to ensure that a channel is still
932            // created with the default uri
933            None => Err(anyhow::anyhow!("")),
934        };
935        let real_channel = match channel {
936            Ok(c) => {
937                log::debug!("Connected via mDNS");
938                c
939            }
940            Err(e) => {
941                if attempting_mdns {
942                    // mDNS found the robot but the connection failed — don't fall through to
943                    // the remote/signaling URI here.  The parallel `without_mdns` branch in
944                    // `connect()` is already handling that case.  Returning an error lets the
945                    // select! loop surface whichever branch succeeds, and avoids a spurious
946                    // auth attempt against app.viam.com when the device is offline.
947                    log::debug!("Unable to connect via mDNS. Error: {e:#}");
948                    return Err(e);
949                }
950                Self::create_channel(allow_downgrade, &domain, uri_for_auth, false).await?
951            }
952        };
953
954        log::debug!("{}", log_prefixes::ACQUIRING_AUTH_TOKEN);
955        let token = get_auth_token(
956            &mut real_channel.clone(),
957            self.config
958                .credentials
959                .as_ref()
960                .unwrap()
961                .credentials
962                .clone(),
963            self.config
964                .credentials
965                .unwrap()
966                .entity
967                .unwrap_or_else(|| domain.clone()),
968        )
969        .await?;
970        log::debug!("{}", log_prefixes::ACQUIRED_AUTH_TOKEN);
971
972        let channel = ServiceBuilder::new()
973            .layer(AddAuthorizationLayer::bearer(&token))
974            .layer(SetRequestHeaderLayer::overriding(
975                HeaderName::from_static("rpc-host"),
976                HeaderValue::from_str(domain.as_str())?,
977            ))
978            .service(real_channel);
979
980        // TODO (RSDK-14026): support WebRTC over mDNS for offline connections (e.g. video streaming).
981        if disable_webrtc || attempting_mdns {
982            log::debug!("Connected via gRPC");
983            Ok(ViamChannel::DirectPreAuthorized(channel))
984        } else {
985            match maybe_connect_via_webrtc(original_uri, channel.clone(), webrtc_options).await {
986                Ok(webrtc_channel) => Ok(ViamChannel::WebRTC(webrtc_channel)),
987                Err(e) => {
988                    log::error!(
989                    "Unable to establish webrtc connection due to error: [{e}]. Attempting direct connection."
990                );
991                    log::debug!("Connected via gRPC");
992                    Ok(ViamChannel::DirectPreAuthorized(channel))
993                }
994            }
995        }
996    }
997
998    async fn connect_mdns(self, original_uri: Parts) -> Result<ViamChannel> {
999        // NOTE(benjirewis): Use a duration of 1500ms for getting the mDNS URI. I've anecdotally
1000        // seen times as great as 922ms to fetch a non-loopback mDNS URI. With an
1001        // interface_with_loopback query interval of 250ms, 1500ms here should give us time for ~6
1002        // queries.
1003        let mdns_uri =
1004            webrtc::action_with_timeout(self.get_mdns_uri(), Duration::from_millis(1500))
1005                .await
1006                .ok()
1007                .flatten()
1008                .ok_or(anyhow::anyhow!(
1009                    "Unable to establish connection via mDNS; uri not found"
1010                ))?;
1011
1012        self.connect_inner(Some(mdns_uri), original_uri).await
1013    }
1014
1015    /// attempts to establish a connection with credentials to the DialBuilder's given uri
1016    pub async fn connect(self) -> Result<ViamChannel> {
1017        log::debug!("{}", log_prefixes::DIAL_ATTEMPT);
1018        let original_uri = self.duplicate_uri().ok_or(anyhow::anyhow!(
1019            "Attempting to connect but there was no uri"
1020        ))?;
1021        let original_uri2 = duplicate_uri(&original_uri).ok_or(anyhow::anyhow!(
1022            "Attempting to connect but there was no uri"
1023        ))?;
1024
1025        let skip_mdns = self.config.disable_mdns;
1026
1027        // We want to short circuit and return the first `Ok` result from our connection
1028        // attempts, which `tokio::select!` does great. Buuuuut, we don't want to
1029        // abandon the `Err` results, and we want to provide comprehensive logging for
1030        // debugging purposes. Hence the loop and pinning. The pinning lets us reference
1031        // the same future multiple times, while the loop lets us immediately return on the
1032        // first `Ok` result while still seeing and logging any error results.
1033        //
1034        // When mDNS is skipped (disable_mdns), with_mdns_err is pre-set so
1035        // the select guard disables that branch and only the direct connection is attempted.
1036        tokio::pin! {
1037            let with_mdns = self.clone().connect_mdns(original_uri);
1038            let without_mdns = self.connect_inner(None, original_uri2);
1039        }
1040        let mut with_mdns_err: Option<anyhow::Error> =
1041            skip_mdns.then(|| anyhow::anyhow!("mDNS skipped"));
1042        let mut without_mdns_err: Option<anyhow::Error> = None;
1043        while with_mdns_err.is_none() || without_mdns_err.is_none() {
1044            tokio::select! {
1045                with_mdns = &mut with_mdns, if with_mdns_err.is_none() => {
1046                    match with_mdns {
1047                        Ok(chan) => return Ok(chan),
1048                        Err(e) => {
1049                            log::debug!("Error connecting with mdns: {e}");
1050                            with_mdns_err = Some(e);
1051                        }
1052                    }
1053                }
1054                without_mdns = &mut without_mdns, if without_mdns_err.is_none() => {
1055                    match without_mdns {
1056                        Ok(chan) => return Ok(chan),
1057                        Err(e) => {
1058                            log::debug!("Error connecting without mdns: {e}");
1059                            without_mdns_err = Some(e);
1060                        }
1061                    }
1062                }
1063            }
1064        }
1065        Err(anyhow::anyhow!(
1066            "Unable to connect with or without mdns.
1067                    with_mdns err: {with_mdns_err:?}
1068                    without_mdns err: {without_mdns_err:?}"
1069        ))
1070    }
1071}
1072
1073async fn send_done_or_error_update(
1074    update: CallUpdateRequest,
1075    channel: AddAuthorization<SetRequestHeader<Channel, HeaderValue>>,
1076) {
1077    let mut signaling_client = SignalingServiceClient::new(channel.clone());
1078
1079    if let Err(e) = signaling_client
1080        .call_update(update)
1081        .await
1082        .map_err(anyhow::Error::from)
1083        .map(|_| ())
1084    {
1085        log::error!("Error sending done or error update: {e}")
1086    }
1087}
1088
1089async fn send_error_once(
1090    sent_error: Arc<AtomicBool>,
1091    uuid: &String,
1092    err: &anyhow::Error,
1093    channel: AddAuthorization<SetRequestHeader<Channel, HeaderValue>>,
1094) {
1095    if sent_error.load(Ordering::Acquire) {
1096        return;
1097    }
1098
1099    let err = google::rpc::Status {
1100        code: google::rpc::Code::Unknown.into(),
1101        message: err.to_string(),
1102        details: Vec::new(),
1103    };
1104    sent_error.store(true, Ordering::Release);
1105    let update_request = CallUpdateRequest {
1106        uuid: uuid.to_string(),
1107        update: Some(Update::Error(err)),
1108    };
1109
1110    send_done_or_error_update(update_request, channel).await
1111}
1112
1113async fn send_done_once(
1114    sent_done: Arc<AtomicBool>,
1115    uuid: &String,
1116    channel: AddAuthorization<SetRequestHeader<Channel, HeaderValue>>,
1117) {
1118    if sent_done.load(Ordering::Acquire) {
1119        return;
1120    }
1121    sent_done.store(true, Ordering::Release);
1122    let update_request = CallUpdateRequest {
1123        uuid: uuid.to_string(),
1124        update: Some(Update::Done(true)),
1125    };
1126
1127    send_done_or_error_update(update_request, channel).await
1128}
1129
1130#[derive(Default)]
1131struct CallerUpdateStats {
1132    count: u128,
1133    total_duration: Duration,
1134    max_duration: Duration,
1135}
1136
1137impl fmt::Display for CallerUpdateStats {
1138    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
1139        let average_duration = &self.total_duration.as_millis() / &self.count;
1140        writeln!(
1141            f,
1142            "Caller update statistics: num_updates: {}, average_duration: {}ms, max_duration: {}ms",
1143            &self.count,
1144            average_duration,
1145            &self.max_duration.as_millis()
1146        )?;
1147        Ok(())
1148    }
1149}
1150
1151async fn maybe_connect_via_webrtc(
1152    uri: Uri,
1153    channel: AddAuthorization<SetRequestHeader<Channel, HeaderValue>>,
1154    webrtc_options: Option<Options>,
1155) -> Result<Arc<WebRTCClientChannel>> {
1156    let webrtc_options = webrtc_options.unwrap_or_else(|| Options::infer_from_uri(uri.clone()));
1157    let mut signaling_client = SignalingServiceClient::new(channel.clone());
1158    let response = match signaling_client
1159        .optional_web_rtc_config(OptionalWebRtcConfigRequest::default())
1160        .await
1161    {
1162        Ok(resp) => resp,
1163        Err(e) => {
1164            if e.code() == tonic::Code::Unimplemented {
1165                tonic::Response::new(OptionalWebRtcConfigResponse::default())
1166            } else {
1167                return Err(anyhow::anyhow!(e));
1168            }
1169        }
1170    };
1171
1172    let optional_config = response.into_inner().config;
1173
1174    if webrtc_options.force_relay && webrtc_options.force_p2p {
1175        log::warn!(
1176            "force_relay and force_p2p are both set; forceP2P strips TURN servers that forceRelay requires so the connection will fail");
1177    }
1178
1179    let (base_config, optional_config) = webrtc::apply_ice_policy(
1180        webrtc_options.config,
1181        optional_config,
1182        webrtc_options.force_relay,
1183        webrtc_options.force_p2p,
1184    );
1185
1186    if webrtc_options.force_relay {
1187        log::debug!("force relay enabled; using relay-only ICE transport policy");
1188    }
1189
1190    if webrtc_options.force_p2p {
1191        log::debug!(
1192            "force P2P enabled; stripping TURN servers and ignoring signaling server ICE config"
1193        );
1194    }
1195
1196    let mut config = webrtc::extend_webrtc_config(base_config, optional_config);
1197
1198    if webrtc_options.force_p2p && webrtc_options.turn_uri.is_some() {
1199        log::warn!("force_p2p is set alongside turn_uri; the TURN filter will have no effect since TURN servers were already stripped");
1200    }
1201    let turn_uri = webrtc_options.turn_uri.as_deref().and_then(|s| {
1202        let parsed = webrtc::TurnUri::parse(s);
1203        if parsed.is_none() {
1204            log::warn!("Failed to parse turn_uri, ignoring: {s:?}");
1205        }
1206        parsed
1207    });
1208    config = webrtc::apply_turn_options(config, turn_uri.as_ref());
1209    if let Some(ref uri) = turn_uri {
1210        log::debug!("TURN filter options set: turn_uri={uri:?}");
1211    }
1212
1213    let (peer_connection, data_channel) =
1214        webrtc::new_peer_connection_for_client(config, webrtc_options.disable_trickle_ice).await?;
1215
1216    let sent_done_or_error = Arc::new(AtomicBool::new(false));
1217    let uuid_lock = Arc::new(RwLock::new("".to_string()));
1218    let uuid_for_ice_gathering_thread = uuid_lock.clone();
1219
1220    // Using an mpsc channel to report unrecoverable errors during Signaling, so we
1221    // don't have to wait until the timeout expires before giving up on this attempt.
1222    // The size of the channel is set to 1 since any error (or success) should terminate the function
1223    let (is_open_s, mut is_open_r) = mpsc::channel(1);
1224    let on_open_is_open = is_open_s.clone();
1225
1226    data_channel.on_open(Box::new(move || {
1227        let _ = on_open_is_open.try_send(None); // ignore sending errors, either an error (or success) was already sent or the operation will succeed
1228        Box::pin(async move {})
1229    }));
1230
1231    let exchange_done = Arc::new(AtomicBool::new(false));
1232    let (remote_description_set_s, remote_description_set_r) = watch::channel(None);
1233    let ice_done = Arc::new(tokio::sync::Notify::new());
1234    let ice_done2 = ice_done.clone();
1235    let caller_update_stats = Arc::new(Mutex::new(CallerUpdateStats::default()));
1236
1237    if !webrtc_options.disable_trickle_ice {
1238        let offer = peer_connection.create_offer(None).await?;
1239        let channel2 = channel.clone();
1240        let uuid_lock2 = uuid_lock.clone();
1241        let sent_done_or_error2 = sent_done_or_error.clone();
1242
1243        let exchange_done = exchange_done.clone();
1244
1245        let on_local_ice_candidate_failure = is_open_s.clone();
1246
1247        let caller_update_stats = caller_update_stats.clone();
1248        let caller_update_stats2 = caller_update_stats.clone();
1249        peer_connection.on_ice_connection_state_change(Box::new(
1250            move |state: RTCIceConnectionState| {
1251                let caller_update_stats = caller_update_stats.clone();
1252                Box::pin(async move {
1253                    if state == RTCIceConnectionState::Completed {
1254                        let caller_update_stats_inner = caller_update_stats.lock().unwrap();
1255                        log::debug!("{}", caller_update_stats_inner);
1256                    }
1257                })
1258            },
1259        ));
1260        peer_connection.on_ice_candidate(Box::new(
1261            move |ice_candidate: Option<RTCIceCandidate>| {
1262                if exchange_done.load(Ordering::Acquire) {
1263                    return Box::pin(async move {});
1264                }
1265                let channel = channel2.clone();
1266                let sent_done_or_error = sent_done_or_error2.clone();
1267                let ice_done = ice_done.clone();
1268                let uuid_lock = uuid_lock2.clone();
1269                let on_local_ice_candidate_failure = on_local_ice_candidate_failure.clone();
1270                let mut remote_description_set_r = remote_description_set_r.clone();
1271                let caller_update_stats = caller_update_stats2.clone();
1272                Box::pin(async move {
1273                    // If the value in the watch channel has not been set yet, we wait until it does.
1274                    // Afterwards Some(()) should be visible to all watcher and any watcher waiting  will
1275                    // return
1276                    if remote_description_set_r.borrow().is_none() {
1277                        match webrtc_action_with_timeout(remote_description_set_r.changed()).await {
1278                            Ok(Err(e)) => {
1279                                let _ = on_local_ice_candidate_failure.try_send(Some(Box::new(
1280                                    anyhow::anyhow!(
1281                                        "remote description watch channel is closed with error {e}"
1282                                    ),
1283                                )));
1284                            }
1285                            Err(_) => {
1286                                log::info!(
1287                                    "timed out on_ice_candidate; remote description was never set"
1288                                );
1289                                let _ = on_local_ice_candidate_failure.try_send(Some(Box::new(
1290                                    anyhow::anyhow!("timed out waiting for remote description"),
1291                                )));
1292                            }
1293                            _ => (),
1294                        }
1295                    }
1296
1297                    let uuid = uuid_lock.read().unwrap().to_string();
1298                    // Note(ethan): for reasons that aren't entirely clear to me, parallel dialing
1299                    // occasionally causes us to not receive a signaling client response when
1300                    // trying to establish a connection. This results in noisy error messages that
1301                    // fortunately are harmless (this problem seems to only ever affect one branch
1302                    // of the parallel dial, so we still end up with a successful connection).
1303                    // By checking if the `uuid` is empty, we can tell if we're in such a case and
1304                    // exit out before it results in logging noisy error messages.
1305                    //
1306                    // It would be lovely to understand this problem better, but given that it's
1307                    // not actually causing performance failures it's probably not worth the effort
1308                    // at this time.
1309                    if uuid.is_empty() {
1310                        log::debug!(
1311                            "UUID never updated. This is likely because we never received a response \
1312                            from the signaling client. This happens occasionally with parallel dialing \
1313                            and isn't concerning provided connection still occurs."
1314                        );
1315                        return;
1316                    }
1317                    let mut signaling_client = SignalingServiceClient::new(channel.clone());
1318                    match ice_candidate {
1319                        Some(ice_candidate) => {
1320                            log::debug!("Gathered local candidate of {ice_candidate}");
1321                            if sent_done_or_error.load(Ordering::Acquire) {
1322                                return;
1323                            }
1324                            let proto_candidate = ice_candidate_to_proto(ice_candidate).await;
1325                            match proto_candidate {
1326                                Ok(proto_candidate) => {
1327                                    let update_request = CallUpdateRequest {
1328                                        uuid: uuid.clone(),
1329                                        update: Some(Update::Candidate(proto_candidate)),
1330                                    };
1331                                    let call_update_start = Instant::now();
1332                                    if let Err(e) = webrtc_action_with_timeout(
1333                                        signaling_client.call_update(update_request),
1334                                    )
1335                                    .await
1336                                    .and_then(|resp| resp.map_err(anyhow::Error::from))
1337                                    {
1338                                        log::error!("Error sending ice candidate: {e}");
1339                                        let _ = on_local_ice_candidate_failure.try_send(Some(
1340                                            Box::new(anyhow::anyhow!(
1341                                                "Error sending ice candidate: {e}"
1342                                            )),
1343                                        ));
1344                                    }
1345                                    let mut caller_update_stats_inner =
1346                                        caller_update_stats.lock().unwrap();
1347                                    caller_update_stats_inner.count += 1;
1348                                    let call_update_duration = call_update_start.elapsed();
1349                                    if call_update_duration > caller_update_stats_inner.max_duration
1350                                    {
1351                                        caller_update_stats_inner.max_duration =
1352                                            call_update_duration;
1353                                    }
1354                                    caller_update_stats_inner.total_duration +=
1355                                        call_update_duration;
1356                                }
1357                                Err(e) => log::error!("Error parsing ice candidate: {e}"),
1358                            }
1359                        }
1360                        None => {
1361                            // will only be executed once when gathering is finished
1362                            ice_done.notify_one();
1363                            send_done_once(sent_done_or_error, &uuid, channel.clone()).await;
1364                        }
1365                    }
1366                })
1367            },
1368        ));
1369
1370        peer_connection.set_local_description(offer).await?;
1371    }
1372
1373    let local_description = peer_connection.local_description().await.unwrap();
1374
1375    // Local SD will be multi-line, so use two log messages to indicate start, SD and end.
1376    log::debug!(
1377        "{}\n{}",
1378        log_prefixes::START_LOCAL_SESSION_DESCRIPTION,
1379        local_description.sdp
1380    );
1381    log::debug!("{}", log_prefixes::END_LOCAL_SESSION_DESCRIPTION);
1382
1383    let sdp = encode_sdp(local_description)?;
1384    let call_request = CallRequest {
1385        sdp,
1386        disable_trickle: webrtc_options.disable_trickle_ice,
1387    };
1388
1389    let client_channel = WebRTCClientChannel::new(peer_connection, data_channel).await;
1390    let client_channel_for_ice_gathering_thread = Arc::downgrade(&client_channel);
1391    let mut signaling_client = SignalingServiceClient::new(channel.clone());
1392    let mut call_client = signaling_client.call(call_request).await?.into_inner();
1393
1394    let channel2 = channel.clone();
1395    let sent_done_or_error2 = sent_done_or_error.clone();
1396    tokio::spawn(async move {
1397        let uuid = uuid_for_ice_gathering_thread;
1398        let client_channel = client_channel_for_ice_gathering_thread;
1399        let init_received = AtomicBool::new(false);
1400        let sent_done = sent_done_or_error2;
1401
1402        loop {
1403            let response = match webrtc_action_with_timeout(call_client.message())
1404                .await
1405                .and_then(|resp| resp.map_err(anyhow::Error::from))
1406            {
1407                Ok(cr) => match cr {
1408                    Some(cr) => cr,
1409                    None => {
1410                        // want to delay sending done until we either are actually done, or
1411                        // we hit a timeout
1412                        let _ = webrtc_action_with_timeout(ice_done2.notified()).await;
1413                        let uuid = uuid.read().unwrap().to_string();
1414                        send_done_once(sent_done.clone(), &uuid, channel2.clone()).await;
1415                        break;
1416                    }
1417                },
1418                Err(e) => {
1419                    log::error!("Error processing call response: {e}");
1420                    let _ = is_open_s.try_send(Some(Box::new(e)));
1421                    break;
1422                }
1423            };
1424
1425            match response.stage {
1426                Some(Stage::Init(init)) => {
1427                    if init_received.load(Ordering::Acquire) {
1428                        let uuid = uuid.read().unwrap().to_string();
1429                        let e = anyhow::anyhow!("Init received more than once");
1430                        send_error_once(sent_done.clone(), &uuid, &e, channel2.clone()).await;
1431                        let _ = is_open_s.try_send(Some(Box::new(e)));
1432                        break;
1433                    }
1434                    init_received.store(true, Ordering::Release);
1435                    {
1436                        let mut uuid_s = uuid.write().unwrap();
1437                        uuid_s.clone_from(&response.uuid);
1438                    }
1439
1440                    let answer = match decode_sdp(init.sdp) {
1441                        Ok(a) => a,
1442                        Err(e) => {
1443                            send_error_once(
1444                                sent_done.clone(),
1445                                &response.uuid,
1446                                &e,
1447                                channel2.clone(),
1448                            )
1449                            .await;
1450                            let _ = is_open_s.try_send(Some(Box::new(e)));
1451                            break;
1452                        }
1453                    };
1454                    {
1455                        let cc = match client_channel.upgrade() {
1456                            Some(cc) => cc,
1457                            None => {
1458                                break;
1459                            }
1460                        };
1461                        if let Err(e) = cc
1462                            .base_channel
1463                            .peer_connection
1464                            .set_remote_description(answer)
1465                            .await
1466                        {
1467                            let e = anyhow::Error::from(e);
1468                            send_error_once(
1469                                sent_done.clone(),
1470                                &response.uuid,
1471                                &e,
1472                                channel2.clone(),
1473                            )
1474                            .await;
1475                            let _ = is_open_s.try_send(Some(Box::new(e)));
1476                            break;
1477                        }
1478                    }
1479                    let _ = remote_description_set_s.send_replace(Some(()));
1480                    if webrtc_options.disable_trickle_ice {
1481                        send_done_once(sent_done.clone(), &response.uuid, channel2.clone()).await;
1482                        break;
1483                    }
1484                }
1485
1486                Some(Stage::Update(update)) => {
1487                    let uuid_s = uuid.read().unwrap().to_string();
1488                    if !init_received.load(Ordering::Acquire) {
1489                        let e = anyhow::anyhow!("Got update before init stage");
1490                        send_error_once(sent_done.clone(), &uuid_s, &e, channel2.clone()).await;
1491                        let _ = is_open_s.try_send(Some(Box::new(e)));
1492                        break;
1493                    }
1494
1495                    if response.uuid != *uuid.read().unwrap() {
1496                        let e = anyhow::anyhow!(
1497                            "uuid mismatch: have {}, want {}",
1498                            response.uuid,
1499                            uuid_s,
1500                        );
1501                        send_error_once(sent_done.clone(), &uuid_s, &e, channel2.clone()).await;
1502                        let _ = is_open_s.try_send(Some(Box::new(e)));
1503                        break;
1504                    }
1505                    match ice_candidate_from_proto(update.candidate) {
1506                        Ok(candidate) => {
1507                            let client_channel = match client_channel.upgrade() {
1508                                Some(cc) => cc,
1509                                None => {
1510                                    break;
1511                                }
1512                            };
1513                            log::debug!("Received remote ICE candidate of {candidate:#?}");
1514                            if let Err(e) = client_channel
1515                                .base_channel
1516                                .peer_connection
1517                                .add_ice_candidate(candidate)
1518                                .await
1519                            {
1520                                let e = anyhow::Error::from(e);
1521                                send_error_once(sent_done.clone(), &uuid_s, &e, channel2.clone())
1522                                    .await;
1523                                let _ = is_open_s.try_send(Some(Box::new(e)));
1524                                break;
1525                            }
1526                        }
1527                        Err(e) => log::error!("Error parsing ice candidate: {e}"),
1528                    }
1529                }
1530                None => continue,
1531            }
1532        }
1533    });
1534
1535    // TODO (GOUT-11): create separate authorization if external_auth_addr and/or creds.Type is `Some`
1536
1537    // Delay returning the client channel until data channel is open, so we don't lose messages
1538    let is_open = webrtc_action_with_timeout(is_open_r.recv()).await;
1539    match is_open {
1540        Ok(is_open) => {
1541            if let Some(Some(e)) = is_open {
1542                return Err(anyhow::anyhow!("Couldn't connect to peer with error {e}"));
1543            }
1544        }
1545        Err(_) => {
1546            return Err(anyhow::anyhow!("Timed out opening data channel."));
1547        }
1548    }
1549
1550    exchange_done.store(true, Ordering::Release);
1551    let uuid = uuid_lock.read().unwrap().to_string();
1552    send_done_once(sent_done_or_error, &uuid, channel.clone()).await;
1553    Ok(client_channel)
1554}
1555
1556async fn ice_candidate_to_proto(ice_candidate: RTCIceCandidate) -> Result<IceCandidate> {
1557    let ice_candidate = ice_candidate.to_json()?;
1558    Ok(IceCandidate {
1559        candidate: ice_candidate.candidate,
1560        sdp_mid: ice_candidate.sdp_mid,
1561        sdpm_line_index: ice_candidate.sdp_mline_index.map(u32::from),
1562        username_fragment: ice_candidate.username_fragment,
1563    })
1564}
1565
1566fn ice_candidate_from_proto(proto: Option<IceCandidate>) -> Result<RTCIceCandidateInit> {
1567    match proto {
1568        Some(proto) => {
1569            let proto_sdpm: usize = proto.sdpm_line_index().try_into()?;
1570            let sdp_mline_index: Option<u16> = proto_sdpm.try_into().ok();
1571
1572            Ok(RTCIceCandidateInit {
1573                candidate: proto.candidate.clone(),
1574                sdp_mid: Some(proto.sdp_mid().to_string()),
1575                sdp_mline_index,
1576                username_fragment: Some(proto.username_fragment().to_string()),
1577            })
1578        }
1579        None => Err(anyhow::anyhow!("No ice candidate provided")),
1580    }
1581}
1582
1583fn decode_sdp(sdp: String) -> Result<RTCSessionDescription> {
1584    let sdp = String::from_utf8(base64::decode(sdp)?)?;
1585    Ok(serde_json::from_str::<RTCSessionDescription>(&sdp)?)
1586}
1587
1588fn encode_sdp(sdp: RTCSessionDescription) -> Result<String> {
1589    let sdp = serde_json::to_vec(&sdp)?;
1590    Ok(base64::encode(sdp))
1591}
1592
1593fn infer_remote_uri_from_authority(uri: Uri, override_addr: Option<&str>) -> Uri {
1594    if let Some(addr) = override_addr {
1595        return Uri::from_parts(uri_parts_with_defaults(addr)).unwrap_or_else(|e| {
1596            log::warn!("Failed to parse signaling server override {addr:?}: {e}; falling back to original URI");
1597            uri
1598        });
1599    }
1600    let authority = uri.authority().map(Authority::as_str).unwrap_or_default();
1601    let is_local_connection = authority.contains(".local.viam.cloud")
1602        || authority.contains("localhost")
1603        || authority.contains("0.0.0.0")
1604        || authority.contains("127.0.0.1");
1605
1606    if !is_local_connection {
1607        if let Some((new_uri, _)) = Options::infer_signaling_server_address(&uri) {
1608            return Uri::from_parts(uri_parts_with_defaults(&new_uri)).unwrap_or(uri);
1609        }
1610    }
1611    uri
1612}
1613
1614fn duplicate_uri(parts: &Parts) -> Option<Parts> {
1615    let uri = Uri::builder()
1616        .authority(parts.authority.clone()?)
1617        .path_and_query(parts.path_and_query.clone()?)
1618        .scheme(parts.scheme.clone()?);
1619    Some(uri.build().ok()?.into_parts())
1620}
1621
1622fn uri_parts_with_defaults(uri: &str) -> Parts {
1623    let mut uri_parts = uri.parse::<Uri>().unwrap().into_parts();
1624    uri_parts.scheme = Some(Scheme::HTTPS);
1625    uri_parts.path_and_query = Some(PathAndQuery::from_static(""));
1626    uri_parts
1627}
1628
1629fn metadata_from_parts(parts: &http::request::Parts) -> Metadata {
1630    let mut md = HashMap::new();
1631    for (k, v) in parts.headers.iter() {
1632        let k = k.to_string();
1633        let v = Strings {
1634            values: vec![HeaderValue::to_str(v).unwrap().to_string()],
1635        };
1636        md.insert(k, v);
1637    }
1638    Metadata { md }
1639}
viam_rust_utils/rpc/dial.rs

viam_rust_utils/rpc/
dial.rs
