iroh_services/
client.rs

1use std::{
2    str::FromStr,
3    sync::{Arc, RwLock},
4};
5
6use anyhow::{Result, anyhow, ensure};
7use iroh::{Endpoint, EndpointAddr, EndpointId, endpoint::ConnectError};
8use iroh_metrics::{MetricsGroup, Registry, encoding::Encoder};
9use irpc_iroh::IrohLazyRemoteConnection;
10use n0_error::StackResultExt;
11use n0_future::{task::AbortOnDropHandle, time::Duration};
12use rcan::Rcan;
13use tokio::sync::oneshot;
14use tracing::{debug, trace, warn};
15use uuid::Uuid;
16
17use crate::{
18    api_secret::{API_SECRET_ENV_VAR_NAME, ApiSecret},
19    caps::{Caps, DEFAULT_CAP_EXPIRY},
20    net_diagnostics::{DiagnosticsReport, checks::run_diagnostics},
21    protocol::{
22        ALPN, Auth, IrohServicesClient, NameEndpoint, Ping, Pong, PutMetrics,
23        PutNetworkDiagnostics, RemoteError,
24    },
25};
26
27/// Client is the main handle for interacting with iroh-services. It communicates with
28/// iroh-services entirely through an iroh endpoint, and is configured through a builder.
29/// Client requires either an Ssh Key or [`ApiSecret`]
30///
31/// ```no_run
32/// use iroh::{Endpoint, endpoint::presets};
33/// use iroh_services::Client;
34///
35/// async fn build_client() -> anyhow::Result<()> {
36///     let endpoint = Endpoint::bind(presets::N0).await?;
37///
38///     // needs IROH_SERVICES_API_SECRET set to an environment variable
39///     // client will now push endpoint metrics to iroh-services.
40///     let client = Client::builder(&endpoint)
41///         .api_secret_from_str("MY_API_SECRET")?
42///         .build()
43///         .await;
44///
45///     Ok(())
46/// }
47/// ```
48///
49/// [`ApiSecret`]: crate::api_secret::ApiSecret
50#[derive(Debug, Clone)]
51pub struct Client {
52    // owned clone of the endpoint for diagnostics, and for connection restarts on actor close
53    #[allow(dead_code)]
54    endpoint: Endpoint,
55    message_channel: tokio::sync::mpsc::Sender<ClientActorMessage>,
56    _actor_task: Arc<AbortOnDropHandle<()>>,
57}
58
59/// ClientBuilder provides configures and builds a iroh-services client, typically
60/// created with [`Client::builder`]
61pub struct ClientBuilder {
62    #[allow(dead_code)]
63    cap_expiry: Duration,
64    cap: Option<Rcan<Caps>>,
65    endpoint: Endpoint,
66    name: Option<String>,
67    metrics_interval: Option<Duration>,
68    remote: Option<EndpointAddr>,
69    registry: Registry,
70}
71
72impl ClientBuilder {
73    pub fn new(endpoint: &Endpoint) -> Self {
74        let mut registry = Registry::default();
75        registry.register_all(endpoint.metrics());
76
77        Self {
78            cap: None,
79            cap_expiry: DEFAULT_CAP_EXPIRY,
80            endpoint: endpoint.clone(),
81            name: None,
82            metrics_interval: Some(Duration::from_secs(60)),
83            remote: None,
84            registry,
85        }
86    }
87
88    /// Register a metrics group to forward to iroh-services
89    ///
90    /// The default registered metrics uses only the endpoint
91    pub fn register_metrics_group(mut self, metrics_group: Arc<dyn MetricsGroup>) -> Self {
92        self.registry.register(metrics_group);
93        self
94    }
95
96    /// Set the metrics collection interval
97    ///
98    /// Defaults to enabled, every 60 seconds.
99    pub fn metrics_interval(mut self, interval: Duration) -> Self {
100        self.metrics_interval = Some(interval);
101        self
102    }
103
104    /// Disable metrics collection.
105    pub fn disable_metrics_interval(mut self) -> Self {
106        self.metrics_interval = None;
107        self
108    }
109
110    /// Set an optional human-readable name for the endpoint the client is
111    /// constructed with, making metrics from this endpoint easier to identify.
112    /// This is often used for associating with other services in your app,
113    /// like a database user id, machine name, permanent username, etc.
114    ///
115    /// When this builder method is called, the provided name is sent after the
116    /// client initially authenticates the endpoint server-side.
117    /// Errors will not interrupt client construction, instead producing a
118    /// warn-level log. For explicit error handling, use [`Client::set_name`].
119    ///
120    /// names can be any UTF-8 string, with a min length of 2 bytes, and
121    /// maximum length of 128 bytes. **name uniqueness is not enforced
122    /// server-side**, which means using the same name for different endpoints
123    /// will not produce an error
124    pub fn name(mut self, name: impl Into<String>) -> Result<Self> {
125        let name = name.into();
126        validate_name(&name).map_err(BuildError::InvalidName)?;
127        self.name = Some(name);
128        Ok(self)
129    }
130
131    /// Check IROH_SERVICES_API_SECRET environment variable for a valid API secret
132    pub fn api_secret_from_env(self) -> Result<Self> {
133        let ticket = ApiSecret::from_env_var(API_SECRET_ENV_VAR_NAME)?;
134        self.api_secret(ticket)
135    }
136
137    /// set client API secret from an encoded string
138    pub fn api_secret_from_str(self, secret_key: &str) -> Result<Self> {
139        let key = ApiSecret::from_str(secret_key).context("invalid iroh services api secret")?;
140        self.api_secret(key)
141    }
142
143    /// Use a shared secret & remote iroh-services endpoint ID contained within a ticket
144    /// to construct a iroh-services client. The resulting client will have "Client"
145    /// capabilities.
146    ///
147    /// API secrets include remote details within them, and will set both the
148    /// remote and rcan values on the builder
149    pub fn api_secret(mut self, ticket: ApiSecret) -> Result<Self> {
150        let local_id = self.endpoint.id();
151        let rcan = crate::caps::create_api_token_from_secret_key(
152            ticket.secret,
153            local_id,
154            self.cap_expiry,
155            Caps::for_shared_secret(),
156        )?;
157
158        self.remote = Some(ticket.remote);
159        self.rcan(rcan)
160    }
161
162    /// Loads the private ssh key from the given path, and creates the needed capability.
163    ///
164    /// The file must contain an unencrypted PEM-encoded OpenSSH ed25519 private key.
165    #[cfg(not(wasm_browser))]
166    pub async fn ssh_key_from_file<P: AsRef<std::path::Path>>(self, path: P) -> Result<Self> {
167        let file_content = tokio::fs::read_to_string(path).await?;
168        self.ssh_key(&file_content)
169    }
170
171    /// Creates the capability from the provided PEM-encoded OpenSSH ed25519 private key.
172    #[cfg(not(wasm_browser))]
173    pub fn ssh_key(mut self, pem: &str) -> Result<Self> {
174        let local_id = self.endpoint.id();
175        let rcan = crate::caps::create_api_token_from_openssh_pem(
176            pem,
177            local_id,
178            self.cap_expiry,
179            Caps::all(),
180        )?;
181        self.cap.replace(rcan);
182
183        Ok(self)
184    }
185
186    /// Sets the rcan directly.
187    pub fn rcan(mut self, cap: Rcan<Caps>) -> Result<Self> {
188        ensure!(
189            EndpointId::from_verifying_key(*cap.audience()) == self.endpoint.id(),
190            "invalid audience"
191        );
192        self.cap.replace(cap);
193        Ok(self)
194    }
195
196    /// Sets the remote to dial, must be provided either directly by calling
197    /// this method, or through calling the api_secret builder methods.
198    pub fn remote(mut self, remote: impl Into<EndpointAddr>) -> Self {
199        self.remote = Some(remote.into());
200        self
201    }
202
203    /// Create a new client, connected to the provide service node
204    #[must_use = "dropping the client will silently cancel all client tasks"]
205    pub async fn build(self) -> Result<Client, BuildError> {
206        debug!("starting iroh-services client");
207        let remote = self.remote.ok_or(BuildError::MissingRemote)?;
208        let capabilities = self.cap.ok_or(BuildError::MissingCapability)?;
209
210        let conn = IrohLazyRemoteConnection::new(self.endpoint.clone(), remote, ALPN.to_vec());
211        let irpc_client = IrohServicesClient::boxed(conn);
212
213        let registry = Arc::new(RwLock::new(self.registry));
214        let (tx, rx) = tokio::sync::mpsc::channel(1);
215        let actor_task = AbortOnDropHandle::new(n0_future::task::spawn(
216            ClientActor {
217                capabilities,
218                client: irpc_client,
219                name: self.name.clone(),
220                session_id: Uuid::new_v4(),
221                authorized: false,
222                encoder: Encoder::new(registry.clone()),
223                registry,
224            }
225            .run(self.name, self.metrics_interval, rx),
226        ));
227
228        Ok(Client {
229            endpoint: self.endpoint,
230            message_channel: tx,
231            _actor_task: Arc::new(actor_task),
232        })
233    }
234}
235
236#[derive(thiserror::Error, Debug)]
237pub enum BuildError {
238    #[error("Missing remote endpoint to dial")]
239    MissingRemote,
240    #[error("Missing capability")]
241    MissingCapability,
242    #[error("Unauthorized")]
243    Unauthorized,
244    #[error("Remote error: {0}")]
245    Remote(#[from] RemoteError),
246    #[error("Rpc connection error: {0}")]
247    Rpc(irpc::Error),
248    #[error("Connection error: {0}")]
249    Connect(ConnectError),
250    #[error("Invalid endpoint name: {0}")]
251    InvalidName(#[from] ValidateNameError),
252}
253
254impl From<irpc::Error> for BuildError {
255    fn from(value: irpc::Error) -> Self {
256        match value {
257            irpc::Error::Request {
258                source:
259                    irpc::RequestError::Connection {
260                        source: iroh::endpoint::ConnectionError::ApplicationClosed(frame),
261                        ..
262                    },
263                ..
264            } if frame.error_code == 401u32.into() => Self::Unauthorized,
265            value => Self::Rpc(value),
266        }
267    }
268}
269
270/// Minimum length in bytes for an endpoint name.
271pub const CLIENT_NAME_MIN_LENGTH: usize = 2;
272/// Maximum length in bytes for an endpoint name.
273pub const CLIENT_NAME_MAX_LENGTH: usize = 128;
274
275/// Error returned when an endpoint name fails validation.
276#[derive(Debug, thiserror::Error)]
277pub enum ValidateNameError {
278    #[error("Name is too long (must be no more than {CLIENT_NAME_MAX_LENGTH} characters).")]
279    TooLong,
280    #[error("Name is too short (must be at least {CLIENT_NAME_MIN_LENGTH} characters).")]
281    TooShort,
282}
283
284fn validate_name(name: &str) -> Result<(), ValidateNameError> {
285    if name.len() < CLIENT_NAME_MIN_LENGTH {
286        Err(ValidateNameError::TooShort)
287    } else if name.len() > CLIENT_NAME_MAX_LENGTH {
288        Err(ValidateNameError::TooLong)
289    } else {
290        Ok(())
291    }
292}
293
294#[derive(thiserror::Error, Debug)]
295pub enum Error {
296    #[error("Invalid endpoint name: {0}")]
297    InvalidName(#[from] ValidateNameError),
298    #[error("Remote error: {0}")]
299    Remote(#[from] RemoteError),
300    #[error("Connection error: {0}")]
301    Rpc(#[from] irpc::Error),
302    #[error(transparent)]
303    Other(#[from] anyhow::Error),
304}
305
306impl Client {
307    pub fn builder(endpoint: &Endpoint) -> ClientBuilder {
308        ClientBuilder::new(endpoint)
309    }
310
311    /// Read the current endpoint name from the local client.
312    pub async fn name(&self) -> Result<Option<String>, Error> {
313        let (tx, rx) = oneshot::channel();
314        self.message_channel
315            .send(ClientActorMessage::ReadName { done: tx })
316            .await
317            .map_err(|_| Error::Other(anyhow!("sending name read request")))?;
318
319        rx.await
320            .map_err(|e| Error::Other(anyhow!("response on internal channel: {:?}", e)))
321    }
322
323    /// Name the active endpoint cloud-side.
324    ///
325    /// names can be any UTF-8 string, with a min length of 2 bytes, and
326    /// maximum length of 128 bytes. **name uniqueness is not enforced.**
327    pub async fn set_name(&self, name: impl Into<String>) -> Result<(), Error> {
328        set_name_inner(self.message_channel.clone(), name.into()).await
329    }
330
331    /// Pings the remote node.
332    pub async fn ping(&self) -> Result<Pong, Error> {
333        let (tx, rx) = oneshot::channel();
334        self.message_channel
335            .send(ClientActorMessage::Ping { done: tx })
336            .await
337            .map_err(|_| Error::Other(anyhow!("sending ping request")))?;
338
339        rx.await
340            .map_err(|e| Error::Other(anyhow!("response on internal channel: {:?}", e)))?
341            .map_err(Error::Remote)
342    }
343
344    /// immediately send a single dump of metrics to iroh-services. It's not necessary
345    /// to call this function if you're using a non-zero metrics interval,
346    /// which will automatically propagate metrics on the set interval for you
347    pub async fn push_metrics(&self) -> Result<(), Error> {
348        let (tx, rx) = oneshot::channel();
349        self.message_channel
350            .send(ClientActorMessage::SendMetrics { done: tx })
351            .await
352            .map_err(|_| Error::Other(anyhow!("sending metrics")))?;
353
354        rx.await
355            .map_err(|e| Error::Other(anyhow!("response on internal channel: {:?}", e)))?
356            .map_err(Error::Remote)
357    }
358
359    /// Grant capabilities to a remote endpoint. Creates a signed RCAN token
360    /// and sends it to iroh-services for storage. The remote can then use this token
361    /// when dialing back to authorize its requests.
362    pub async fn grant_capability(
363        &self,
364        remote_id: EndpointId,
365        caps: impl IntoIterator<Item = impl Into<crate::caps::Cap>>,
366    ) -> Result<(), Error> {
367        let cap = crate::caps::create_grant_token(
368            self.endpoint.secret_key().clone(),
369            remote_id,
370            DEFAULT_CAP_EXPIRY,
371            Caps::new(caps),
372        )
373        .map_err(Error::Other)?;
374
375        let (tx, rx) = oneshot::channel();
376        self.message_channel
377            .send(ClientActorMessage::GrantCap {
378                cap: Box::new(cap),
379                done: tx,
380            })
381            .await
382            .map_err(|_| Error::Other(anyhow!("granting capability")))?;
383
384        rx.await
385            .map_err(|e| Error::Other(anyhow!("response on internal channel: {:?}", e)))?
386    }
387
388    /// run local network status diagnostics, optionally uploading the results
389    pub async fn net_diagnostics(&self, send: bool) -> Result<DiagnosticsReport, Error> {
390        let report = run_diagnostics(&self.endpoint).await?;
391        if send {
392            let (tx, rx) = oneshot::channel();
393            self.message_channel
394                .send(ClientActorMessage::PutNetworkDiagnostics {
395                    done: tx,
396                    report: Box::new(report.clone()),
397                })
398                .await
399                .map_err(|_| Error::Other(anyhow!("sending network diagnostics report")))?;
400
401            let _ = rx
402                .await
403                .map_err(|e| Error::Other(anyhow!("response on internal channel: {:?}", e)))?;
404        }
405
406        Ok(report)
407    }
408}
409
410enum ClientActorMessage {
411    SendMetrics {
412        done: oneshot::Sender<Result<(), RemoteError>>,
413    },
414    Ping {
415        done: oneshot::Sender<Result<Pong, RemoteError>>,
416    },
417    // GrantCap is used by the `client_host` feature flag
418    #[allow(dead_code)]
419    GrantCap {
420        // boxed to avoid large enum variants
421        cap: Box<Rcan<Caps>>,
422        done: oneshot::Sender<Result<(), Error>>,
423    },
424    PutNetworkDiagnostics {
425        report: Box<DiagnosticsReport>,
426        done: oneshot::Sender<Result<(), Error>>,
427    },
428    ReadName {
429        done: oneshot::Sender<Option<String>>,
430    },
431    NameEndpoint {
432        name: String,
433        done: oneshot::Sender<Result<(), RemoteError>>,
434    },
435}
436
437struct ClientActor {
438    capabilities: Rcan<Caps>,
439    client: IrohServicesClient,
440    name: Option<String>,
441    session_id: Uuid,
442    authorized: bool,
443    encoder: Encoder,
444    // Kept so auth() can rebuild the encoder; a fresh encoder re-sends the schema.
445    registry: Arc<RwLock<Registry>>,
446}
447
448impl ClientActor {
449    async fn run(
450        mut self,
451        initial_name: Option<String>,
452        interval: Option<Duration>,
453        mut inbox: tokio::sync::mpsc::Receiver<ClientActorMessage>,
454    ) {
455        let mut metrics_timer = interval.map(|interval| n0_future::time::interval(interval));
456        trace!("starting client actor");
457
458        if let Some(name) = initial_name
459            && let Err(err) = self.send_name_endpoint(name).await
460        {
461            warn!(err = %err, "failed setting endpoint name on startup");
462        }
463
464        loop {
465            trace!("client actor tick");
466            tokio::select! {
467                biased;
468                Some(msg) = inbox.recv() => {
469                    match msg {
470                        ClientActorMessage::Ping{ done } => {
471                            let res = self.send_ping().await;
472                            if let Err(err) = done.send(res) {
473                                debug!("failed to send ping: {:#?}", err);
474                                self.authorized = false;
475                            }
476                        },
477                        ClientActorMessage::SendMetrics{ done } => {
478                            trace!("sending metrics manually triggered");
479                            let res = self.send_metrics().await;
480                            if let Err(err) = done.send(res) {
481                                debug!("failed to push metrics: {:#?}", err);
482                                self.authorized = false;
483                            }
484                        }
485                        ClientActorMessage::GrantCap{ cap, done } => {
486                            let res = self.grant_cap(*cap).await;
487                            if let Err(err) = done.send(res) {
488                                warn!("failed to grant capability: {:#?}", err);
489                            }
490                        }
491                        ClientActorMessage::ReadName{ done } => {
492                            if let Err(err) = done.send(self.name.clone()) {
493                                warn!("sending name value: {:#?}", err);
494                            }
495                        }
496                        ClientActorMessage::NameEndpoint{ name, done } => {
497                            let res = self.send_name_endpoint(name).await;
498                            if let Err(err) = done.send(res) {
499                                warn!("failed to name endpoint: {:#?}", err);
500                            }
501                        }
502                        ClientActorMessage::PutNetworkDiagnostics{ report, done } => {
503                            let res = self.put_network_diagnostics(*report).await;
504                            if let Err(err) = done.send(res) {
505                                warn!("failed to publish network diagnostics: {:#?}", err);
506                            }
507                        }
508                    }
509                }
510                _ = async {
511                    if let Some(ref mut timer) = metrics_timer {
512                        timer.tick().await;
513                    } else {
514                        std::future::pending::<()>().await;
515                    }
516                } => {
517                    trace!("metrics send tick");
518                    if let Err(err) = self.send_metrics().await {
519                        debug!("failed to push metrics: {:#?}", err);
520                        self.authorized = false;
521                    }
522                },
523            }
524        }
525    }
526
527    // sends an authorization request to the server
528    async fn auth(&mut self) -> Result<(), RemoteError> {
529        if self.authorized {
530            return Ok(());
531        }
532        trace!("client authorizing");
533        self.client
534            .rpc(Auth {
535                caps: self.capabilities.clone(),
536            })
537            .await
538            .inspect_err(|e| debug!("authorization failed: {:?}", e))
539            .map_err(|e| RemoteError::AuthError(e.to_string()))?;
540        self.authorized = true;
541        // A completed handshake means a fresh connection, and the server keeps
542        // one metrics decoder per connection. A new encoder starts at schema
543        // version zero, so the next export carries the full schema for that
544        // decoder. Re-importing a schema is idempotent server-side, so
545        // re-sending after a spurious re-auth is harmless.
546        self.encoder = Encoder::new(self.registry.clone());
547        Ok(())
548    }
549
550    /// Marks the session unauthorized when an rpc fails.
551    ///
552    /// The lazy connection re-dials on the next request, and the server
553    /// accepts only Auth first on a new connection, so the next call must run
554    /// the handshake again (which also re-sends the metrics schema).
555    fn track_rpc<T>(&mut self, res: Result<T, irpc::Error>) -> Result<T, irpc::Error> {
556        if res.is_err() {
557            self.authorized = false;
558        }
559        res
560    }
561
562    async fn send_ping(&mut self) -> Result<Pong, RemoteError> {
563        trace!("client actor send ping");
564        self.auth().await?;
565
566        let req = rand::random();
567        let res = self.client.rpc(Ping { req_id: req }).await;
568        self.track_rpc(res)
569            .inspect_err(|e| warn!("rpc ping error: {e}"))
570            .map_err(|_| RemoteError::InternalServerError)
571    }
572
573    async fn send_name_endpoint(&mut self, name: String) -> Result<(), RemoteError> {
574        trace!("client sending name endpoint request");
575        self.auth().await?;
576
577        let res = self.client.rpc(NameEndpoint { name: name.clone() }).await;
578        self.track_rpc(res)
579            .inspect_err(|e| debug!("name endpoint error: {e}"))
580            .map_err(|_| RemoteError::InternalServerError)??;
581        self.name = Some(name);
582        Ok(())
583    }
584
585    async fn send_metrics(&mut self) -> Result<(), RemoteError> {
586        trace!("client actor send metrics");
587        self.auth().await?;
588
589        let update = self.encoder.export();
590        // let delta = update_delta(&self.latest_ackd_update, &update);
591        let req = PutMetrics {
592            session_id: self.session_id,
593            update,
594        };
595
596        let res = self.client.rpc(req).await;
597        self.track_rpc(res)
598            .map_err(|_| RemoteError::InternalServerError)??;
599
600        Ok(())
601    }
602
603    async fn grant_cap(&mut self, cap: Rcan<Caps>) -> Result<(), Error> {
604        trace!("client actor grant capability");
605        self.auth().await?;
606
607        let res = self.client.rpc(crate::protocol::GrantCap { cap }).await;
608        self.track_rpc(res)
609            .map_err(|_| RemoteError::InternalServerError)??;
610
611        Ok(())
612    }
613
614    async fn put_network_diagnostics(
615        &mut self,
616        report: crate::net_diagnostics::DiagnosticsReport,
617    ) -> Result<(), Error> {
618        trace!("client actor publish network diagnostics");
619        self.auth().await?;
620
621        let req = PutNetworkDiagnostics { report };
622
623        let res = self.client.rpc(req).await;
624        self.track_rpc(res)
625            .map_err(|_| RemoteError::InternalServerError)??;
626
627        Ok(())
628    }
629}
630
631async fn set_name_inner(
632    message_channel: tokio::sync::mpsc::Sender<ClientActorMessage>,
633    name: String,
634) -> Result<(), Error> {
635    validate_name(&name)?;
636    debug!(name_len = name.len(), "calling set name");
637    let (tx, rx) = oneshot::channel();
638    message_channel
639        .send(ClientActorMessage::NameEndpoint { name, done: tx })
640        .await
641        .map_err(|_| Error::Other(anyhow!("sending name endpoint request")))?;
642    rx.await
643        .map_err(|e| Error::Other(anyhow!("response on internal channel: {:?}", e)))?
644        .map_err(Error::Remote)
645}
646
647#[cfg(test)]
648mod tests {
649    use std::sync::{
650        Arc, RwLock,
651        atomic::{AtomicBool, Ordering},
652    };
653
654    use iroh::{
655        Endpoint, EndpointAddr, SecretKey,
656        endpoint::{Connection, presets},
657        protocol::{AcceptError, ProtocolHandler, Router},
658    };
659    use iroh_metrics::{
660        Registry,
661        encoding::{Decoder, Encoder},
662    };
663    use irpc::WithChannels;
664    use irpc_iroh::read_request;
665    use n0_error::AnyError;
666    use n0_future::time::Duration;
667    use rand::{RngExt, SeedableRng};
668    use temp_env_vars::temp_env_vars;
669
670    use crate::{
671        Client,
672        api_secret::ApiSecret,
673        caps::{Cap, Caps, create_api_token_from_secret_key},
674        client::{API_SECRET_ENV_VAR_NAME, BuildError, ValidateNameError},
675        protocol::{ALPN, IrohServicesProtocol, ServicesMessage},
676    };
677
678    /// What the test server recorded about one PutMetrics request.
679    #[derive(Debug)]
680    struct SeenUpdate {
681        has_schema: bool,
682        decoded_items: usize,
683    }
684
685    /// In-process stand-in for the services backend.
686    ///
687    /// Mirrors the real server's session rules: the first request on every
688    /// connection must be Auth, and the metrics decoder lives per connection.
689    #[derive(Debug)]
690    struct RecordingServer {
691        seen: tokio::sync::mpsc::UnboundedSender<SeenUpdate>,
692        drop_next: Arc<AtomicBool>,
693    }
694
695    impl ProtocolHandler for RecordingServer {
696        async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
697            self.handle_connection(connection).await.map_err(|e| {
698                let boxed: Box<dyn std::error::Error + Send + Sync> = e.into();
699                AcceptError::from(AnyError::from(boxed))
700            })
701        }
702    }
703
704    impl RecordingServer {
705        async fn handle_connection(&self, connection: Connection) -> anyhow::Result<()> {
706            let mut decoder = Decoder::default();
707            let mut authed = false;
708            loop {
709                let Some(request) = read_request::<IrohServicesProtocol>(&connection).await? else {
710                    return Ok(());
711                };
712                if self.drop_next.swap(false, Ordering::SeqCst) {
713                    // Simulates a server restart: the connection dies without an
714                    // answer and takes the per-connection decoder with it.
715                    connection.close(500u32.into(), b"test restart");
716                    return Ok(());
717                }
718                match request {
719                    ServicesMessage::Auth(WithChannels { tx, .. }) => {
720                        authed = true;
721                        tx.send(()).await?;
722                    }
723                    ServicesMessage::PutMetrics(WithChannels { inner, tx, .. }) if authed => {
724                        let has_schema = inner.update.schema.is_some();
725                        decoder.import(inner.update);
726                        let decoded_items = decoder.iter().count();
727                        let _ = self.seen.send(SeenUpdate {
728                            has_schema,
729                            decoded_items,
730                        });
731                        tx.send(Ok(())).await?;
732                    }
733                    _ => {
734                        connection.close(400u32.into(), b"Expected initial auth message");
735                        return Ok(());
736                    }
737                }
738            }
739        }
740    }
741
742    /// A reconnect must make the next metrics update carry the schema.
743    ///
744    /// The server holds one decoder per connection, so the first update after
745    /// a re-auth is undecodable unless the schema comes along again.
746    #[tokio::test]
747    async fn test_metrics_schema_resent_after_reconnect() {
748        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(2);
749        let server_ep = Endpoint::builder(presets::Minimal).bind().await.unwrap();
750        let client_ep = Endpoint::builder(presets::Minimal).bind().await.unwrap();
751
752        let (seen_tx, mut seen_rx) = tokio::sync::mpsc::unbounded_channel();
753        let drop_next = Arc::new(AtomicBool::new(false));
754        let server = RecordingServer {
755            seen: seen_tx,
756            drop_next: drop_next.clone(),
757        };
758        let router = Router::builder(server_ep.clone())
759            .accept(ALPN, server)
760            .spawn();
761
762        // The test server accepts any capability, so a self-issued token works.
763        let shared_secret = SecretKey::from_bytes(&rng.random());
764        let cap = create_api_token_from_secret_key(
765            shared_secret,
766            client_ep.id(),
767            Duration::from_secs(3600),
768            Caps::for_shared_secret(),
769        )
770        .unwrap();
771
772        let client = Client::builder(&client_ep)
773            .disable_metrics_interval()
774            .remote(server_ep.addr())
775            .rcan(cap)
776            .unwrap()
777            .build()
778            .await
779            .unwrap();
780
781        // The first export on a fresh session carries the schema.
782        client.push_metrics().await.unwrap();
783        let first = seen_rx.recv().await.unwrap();
784        assert!(first.has_schema);
785        assert!(first.decoded_items > 0);
786
787        // Steady state suppresses the schema; the connection's decoder
788        // already has it and keeps decoding.
789        client.push_metrics().await.unwrap();
790        let second = seen_rx.recv().await.unwrap();
791        assert!(!second.has_schema);
792        assert!(second.decoded_items > 0);
793
794        // The server drops the connection mid-request: one failed round trip.
795        drop_next.store(true, Ordering::SeqCst);
796        assert!(client.push_metrics().await.is_err());
797
798        // The client re-dials and re-auths; the first update on the new
799        // connection must include the schema for the server's fresh decoder.
800        client.push_metrics().await.unwrap();
801        let third = seen_rx.recv().await.unwrap();
802        assert!(third.has_schema, "schema must be re-sent after a reconnect");
803        assert!(third.decoded_items > 0);
804
805        router.shutdown().await.unwrap();
806        client_ep.close().await;
807    }
808
809    /// Documents the encoder and decoder contract the reconnect fix relies on.
810    #[tokio::test]
811    async fn test_fresh_encoder_resends_schema() {
812        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();
813        let mut registry = Registry::default();
814        registry.register_all(endpoint.metrics());
815        let registry = Arc::new(RwLock::new(registry));
816
817        let mut encoder = Encoder::new(registry.clone());
818        let first = encoder.export();
819        assert!(first.schema.is_some());
820
821        // Steady state omits the schema once it has been exported.
822        let schemaless = encoder.export();
823        assert!(schemaless.schema.is_none());
824
825        // A decoder that never saw the schema decodes such an update to
826        // nothing: this is the server side of the reconnect bug.
827        let mut fresh_decoder = Decoder::default();
828        fresh_decoder.import(schemaless);
829        assert_eq!(fresh_decoder.iter().count(), 0);
830
831        // A new encoder over the same registry starts at schema version zero
832        // and re-publishes the schema, which is what auth() does on re-auth.
833        let mut encoder = Encoder::new(registry);
834        let resent = encoder.export();
835        assert!(resent.schema.is_some());
836        let mut fresh_decoder = Decoder::default();
837        fresh_decoder.import(resent);
838        assert!(fresh_decoder.iter().count() > 0);
839    }
840
841    #[tokio::test]
842    #[temp_env_vars]
843    async fn test_api_key_from_env() {
844        // construct
845        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(0);
846        let shared_secret = SecretKey::from_bytes(&rng.random());
847        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
848        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);
849        unsafe {
850            std::env::set_var(API_SECRET_ENV_VAR_NAME, api_secret.to_string());
851        };
852
853        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();
854
855        let builder = Client::builder(&endpoint).api_secret_from_env().unwrap();
856
857        let fake_endpoint_addr: EndpointAddr = fake_endpoint_id.into();
858        assert_eq!(builder.remote, Some(fake_endpoint_addr));
859
860        // Compare capability fields individually to avoid flaky timestamp
861        // mismatches between the builder's rcan and a freshly-created one.
862        let cap = builder.cap.as_ref().expect("expected capability to be set");
863        assert_eq!(cap.capability(), &Caps::new([Cap::Client]));
864        assert_eq!(cap.audience(), &endpoint.id().as_verifying_key());
865        assert_eq!(cap.issuer(), &shared_secret.public().as_verifying_key());
866    }
867
868    /// Assert that disabling metrics interval can manually send metrics without
869    /// panicking. Metrics sending itself is expected to fail.
870    #[tokio::test]
871    async fn test_no_metrics_interval() {
872        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(1);
873        let shared_secret = SecretKey::from_bytes(&rng.random());
874        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
875        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);
876
877        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();
878
879        let client = Client::builder(&endpoint)
880            .disable_metrics_interval()
881            .api_secret(api_secret)
882            .unwrap()
883            .build()
884            .await
885            .unwrap();
886
887        let err = client.push_metrics().await;
888        assert!(err.is_err());
889    }
890
891    #[tokio::test]
892    async fn test_name() {
893        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(0);
894        let shared_secret = SecretKey::from_bytes(&rng.random());
895        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
896        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);
897
898        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();
899
900        let builder = Client::builder(&endpoint)
901            .name("my-node 👋")
902            .unwrap()
903            .api_secret(api_secret)
904            .unwrap();
905
906        assert_eq!(builder.name, Some("my-node 👋".to_string()));
907
908        let Err(err) = Client::builder(&endpoint).name("a") else {
909            panic!("name should fail for strings under 2 bytes");
910        };
911        assert!(matches!(
912            err.downcast_ref::<BuildError>(),
913            Some(BuildError::InvalidName(ValidateNameError::TooShort))
914        ));
915
916        let too_long_name = "👋".repeat(129);
917        let Err(err) = Client::builder(&endpoint).name(&too_long_name) else {
918            panic!("name should fail for strings over 128 bytes");
919        };
920        assert!(matches!(
921            err.downcast_ref::<BuildError>(),
922            Some(BuildError::InvalidName(ValidateNameError::TooLong))
923        ));
924    }
925}
iroh_services/client.rs

iroh_services/
client.rs
