iroh_services/

client.rs

use std::{
    collections::BTreeMap,
    fmt::Display,
    str::FromStr,
    sync::{Arc, RwLock},
};

use anyhow::{Result, ensure};
use iroh::{
    Endpoint, EndpointAddr, EndpointId,
    endpoint::{ConnectError, Connection},
};
use iroh_metrics::{MetricsGroup, Registry, encoding::Encoder};
use irpc::{Channels, RpcMessage, WithChannels, channel::none::NoReceiver};
use irpc_iroh::IrohRemoteConnection;
use n0_error::StackResultExt;
use n0_future::{
    task::{self, AbortOnDropHandle},
    time::{self, Duration},
};
use rcan::Rcan;
use tokio::sync::oneshot;
use tokio_util::sync::CancellationToken;
use tracing::{debug, trace, warn};
use uuid::Uuid;

use crate::{
    api_secret::{API_SECRET_ENV_VAR_NAME, ApiSecret},
    caps::{Caps, DEFAULT_CAP_EXPIRY},
    net_diagnostics::{DiagnosticsReport, checks::run_diagnostics},
    protocol::{
        ALPN, Auth, GrantCap, IrohServicesClient, IrohServicesProtocol, NameEndpoint, Ping, Pong,
        PutMetrics, PutNetworkDiagnostics, RemoteError, ServicesMessage, SetAttributes, SetGroup,
    },
};

/// Client is the main handle for interacting with iroh-services. It communicates with
/// iroh-services entirely through an iroh endpoint, and is configured through a builder.
/// Client requires either an Ssh Key or [`ApiSecret`]
///
/// ```no_run
/// use iroh::{Endpoint, endpoint::presets};
/// use iroh_services::Client;
///
/// async fn build_client() -> anyhow::Result<()> {
///     let endpoint = Endpoint::bind(presets::N0).await?;
///
///     // needs IROH_SERVICES_API_SECRET set to an environment variable
///     // client will now push endpoint metrics to iroh-services.
///     let client = Client::builder(&endpoint)
///         .api_secret_from_str("MY_API_SECRET")?
///         .build()
///         .await;
///
///     Ok(())
/// }
/// ```
///
/// [`ApiSecret`]: crate::api_secret::ApiSecret
#[derive(Debug, Clone)]
pub struct Client {
    // owned clone of the endpoint for diagnostics, and for connection restarts on actor close
    endpoint: Endpoint,
    message_channel: tokio::sync::mpsc::Sender<ClientActorMessage>,
    /// Cancelled by [`Client::shutdown`] to stop whatever the actor is doing.
    shutdown: CancellationToken,
    _actor_task: Arc<AbortOnDropHandle<()>>,
}

/// ClientBuilder provides configures and builds a iroh-services client, typically
/// created with [`Client::builder`]
pub struct ClientBuilder {
    cap_expiry: Duration,
    cap: Option<Rcan<Caps>>,
    endpoint: Endpoint,
    name: Option<String>,
    group: Option<String>,
    attributes: Option<BTreeMap<String, String>>,
    metrics_interval: Option<Duration>,
    remote: Option<EndpointAddr>,
    registry: Registry,
}

impl ClientBuilder {
    pub fn new(endpoint: &Endpoint) -> Self {
        let mut registry = Registry::default();
        registry.register_all(endpoint.metrics());

        Self {
            cap: None,
            cap_expiry: DEFAULT_CAP_EXPIRY,
            endpoint: endpoint.clone(),
            name: None,
            group: None,
            attributes: None,
            metrics_interval: Some(Duration::from_secs(60)),
            remote: None,
            registry,
        }
    }

    /// Register a metrics group to forward to iroh-services
    ///
    /// The default registered metrics uses only the endpoint
    pub fn register_metrics_group(mut self, metrics_group: Arc<dyn MetricsGroup>) -> Self {
        self.registry.register(metrics_group);
        self
    }

    /// Set the metrics collection interval
    ///
    /// Defaults to enabled, every 60 seconds.
    pub fn metrics_interval(mut self, interval: Duration) -> Self {
        self.metrics_interval = Some(interval);
        self
    }

    /// Disable metrics collection.
    pub fn disable_metrics_interval(mut self) -> Self {
        self.metrics_interval = None;
        self
    }

    /// Set an optional human-readable name for the endpoint, making its metrics
    /// easier to identify.
    ///
    /// Often a database user id, machine name, or other stable identifier from
    /// your application. A name must be 2 to 128 bytes of UTF-8; uniqueness is not
    /// enforced, so different endpoints may share a name.
    ///
    /// Validation errors are returned here. The name is sent to the server after
    /// the client authenticates; a failure to send it at that point is logged at
    /// warn level rather than returned; use [`Client::set_name`] to set it later
    /// with explicit error handling.
    pub fn name(mut self, name: impl Into<String>) -> Result<Self> {
        let name = name.into();
        validate_name(&name).map_err(BuildError::InvalidName)?;
        self.name = Some(name);
        Ok(self)
    }

    /// Attach the endpoint to a single named group when the client first
    /// authenticates.
    ///
    /// A group name must be 2 to 128 bytes of UTF-8. Validation errors are returned
    /// here. The group is sent to the server after the client authenticates; a
    /// failure to send it at that point is logged at warn level rather than
    /// returned; use [`Client::set_group`] to set it later with explicit error
    /// handling.
    pub fn group(mut self, group: impl Into<String>) -> Result<Self> {
        let group = group.into();
        validate_name(&group).map_err(BuildError::InvalidGroup)?;
        self.group = Some(group);
        Ok(self)
    }

    /// Attach arbitrary key-value attributes to the endpoint when the client
    /// first authenticates. Accepts any iterable of `(key, value)` pairs:
    ///
    /// ```no_run
    /// # use iroh::{Endpoint, endpoint::presets};
    /// # use iroh_services::Client;
    /// # async fn example(endpoint: &Endpoint) -> anyhow::Result<()> {
    /// let _ = Client::builder(endpoint).attributes([("env", "prod"), ("region", "us-west")])?;
    /// # Ok(()) }
    /// ```
    ///
    /// Each key must be 2 to 128 bytes of UTF-8; values may be empty and are capped
    /// at 128 bytes; at most 128 entries are allowed. Validation errors are
    /// returned here. The attributes are sent to the server after the client
    /// authenticates; a failure to send them at that point is logged at warn
    /// level rather than returned; use [`Client::set_attributes`] to set them
    /// later with explicit error handling.
    pub fn attributes<I, K, V>(mut self, attrs: I) -> Result<Self>
    where
        I: IntoIterator<Item = (K, V)>,
        K: Into<String>,
        V: Into<String>,
    {
        let collected: BTreeMap<String, String> = attrs
            .into_iter()
            .map(|(k, v)| (k.into(), v.into()))
            .collect();
        validate_attributes(&collected).map_err(BuildError::InvalidAttributes)?;
        self.attributes = Some(collected);
        Ok(self)
    }

    /// Check IROH_SERVICES_API_SECRET environment variable for a valid API secret
    pub fn api_secret_from_env(self) -> Result<Self> {
        let ticket = ApiSecret::from_env_var(API_SECRET_ENV_VAR_NAME)?;
        self.api_secret(ticket)
    }

    /// set client API secret from an encoded string
    pub fn api_secret_from_str(self, secret_key: &str) -> Result<Self> {
        let key = ApiSecret::from_str(secret_key).context("invalid iroh services api secret")?;
        self.api_secret(key)
    }

    /// Use a shared secret & remote iroh-services endpoint ID contained within a ticket
    /// to construct a iroh-services client. The resulting client will have "Client"
    /// capabilities.
    ///
    /// API secrets include remote details within them, and will set both the
    /// remote and rcan values on the builder
    pub fn api_secret(mut self, ticket: ApiSecret) -> Result<Self> {
        let local_id = self.endpoint.id();
        let rcan = crate::caps::create_api_token_from_secret_key(
            ticket.secret,
            local_id,
            self.cap_expiry,
            Caps::for_shared_secret(),
        )?;

        self.remote = Some(ticket.remote);
        self.rcan(rcan)
    }

    /// Loads the private ssh key from the given path, and creates the needed capability.
    ///
    /// The file must contain an unencrypted PEM-encoded OpenSSH ed25519 private key.
    #[cfg(not(wasm_browser))]
    pub async fn ssh_key_from_file<P: AsRef<std::path::Path>>(self, path: P) -> Result<Self> {
        let file_content = tokio::fs::read_to_string(path).await?;
        self.ssh_key(&file_content)
    }

    /// Creates the capability from the provided PEM-encoded OpenSSH ed25519 private key.
    #[cfg(not(wasm_browser))]
    pub fn ssh_key(mut self, pem: &str) -> Result<Self> {
        let local_id = self.endpoint.id();
        let rcan = crate::caps::create_api_token_from_openssh_pem(
            pem,
            local_id,
            self.cap_expiry,
            Caps::all(),
        )?;
        self.cap.replace(rcan);

        Ok(self)
    }

    /// Sets the rcan directly.
    pub fn rcan(mut self, cap: Rcan<Caps>) -> Result<Self> {
        ensure!(
            EndpointId::from_verifying_key(*cap.audience()) == self.endpoint.id(),
            "invalid audience"
        );
        self.cap.replace(cap);
        Ok(self)
    }

    /// Sets the remote to dial, must be provided either directly by calling
    /// this method, or through calling the api_secret builder methods.
    pub fn remote(mut self, remote: impl Into<EndpointAddr>) -> Self {
        self.remote = Some(remote.into());
        self
    }

    /// Create a new client, connected to the provide service node
    #[must_use = "dropping the client will silently cancel all client tasks"]
    pub async fn build(self) -> Result<Client, BuildError> {
        debug!("starting iroh-services client");
        let remote = self.remote.ok_or(BuildError::MissingRemote)?;
        let capabilities = self.cap.ok_or(BuildError::MissingCapability)?;

        let registry = Arc::new(RwLock::new(self.registry));
        let (tx, rx) = tokio::sync::mpsc::channel(1);
        let shutdown = CancellationToken::new();
        let actor_task = AbortOnDropHandle::new(task::spawn(
            ClientActor {
                capabilities,
                endpoint: self.endpoint.clone(),
                remote,
                client: None,
                name: self.name.clone(),
                group: self.group.clone(),
                attributes: self.attributes.clone().unwrap_or_default(),
                session_id: Uuid::new_v4(),
                encoder: Encoder::new(registry.clone()),
                registry,
            }
            .run(self.metrics_interval, rx, shutdown.clone()),
        ));

        Ok(Client {
            endpoint: self.endpoint,
            message_channel: tx,
            shutdown,
            _actor_task: Arc::new(actor_task),
        })
    }
}

#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum BuildError {
    #[error("Missing remote endpoint to dial")]
    MissingRemote,
    #[error("Missing capability")]
    MissingCapability,
    #[error("Unauthorized")]
    Unauthorized,
    #[error("Remote error: {0}")]
    Remote(#[from] RemoteError),
    #[error("Rpc connection error: {0}")]
    Rpc(irpc::Error),
    #[error("Connection error: {0}")]
    Connect(ConnectError),
    #[error("Invalid endpoint name: {0}")]
    InvalidName(#[from] ValidateNameError),
    #[error("Invalid endpoint group: {0}")]
    InvalidGroup(ValidateNameError),
    #[error("Invalid endpoint attributes: {0}")]
    InvalidAttributes(#[from] ValidateAttributesError),
}

impl From<irpc::Error> for BuildError {
    fn from(value: irpc::Error) -> Self {
        match value {
            irpc::Error::Request {
                source:
                    irpc::RequestError::Connection {
                        source: iroh::endpoint::ConnectionError::ApplicationClosed(frame),
                        ..
                    },
                ..
            } if frame.error_code == 401u32.into() => Self::Unauthorized,
            value => Self::Rpc(value),
        }
    }
}

/// How long [`Client::shutdown`] lets a request already in flight finish.
///
/// Dropping a request mid-flight resets its stream, and the server treats that
/// as the connection failing: it stops reading requests and tears the
/// connection down. Letting the request finish keeps the connection usable for
/// the final metrics push. Requests to a responsive server complete in
/// fast, so this only runs out when the server has stopped answering,
/// in which case the final metrics push is then skipped.
const SHUTDOWN_GRACE: Duration = Duration::from_secs(3);

/// How long [`Client::shutdown`] waits for the final metrics push.
///
/// The push goes over an established connection, so it is normally well under
/// this. The bound covers a peer that has gone silent without closing, where
/// the request would otherwise hang until the connection times out.
const SHUTDOWN_FLUSH_TIMEOUT: Duration = Duration::from_secs(2);

/// Minimum length in bytes for an endpoint name.
pub const CLIENT_NAME_MIN_LENGTH: usize = 2;
/// Maximum length in bytes for an endpoint name.
pub const CLIENT_NAME_MAX_LENGTH: usize = 128;

/// Error returned when an endpoint name fails validation.
#[derive(Debug, thiserror::Error)]
pub enum ValidateNameError {
    #[error("Name is too long (must be no more than {CLIENT_NAME_MAX_LENGTH} bytes).")]
    TooLong,
    #[error("Name is too short (must be at least {CLIENT_NAME_MIN_LENGTH} bytes).")]
    TooShort,
}

fn validate_name(name: &str) -> Result<(), ValidateNameError> {
    if name.len() < CLIENT_NAME_MIN_LENGTH {
        Err(ValidateNameError::TooShort)
    } else if name.len() > CLIENT_NAME_MAX_LENGTH {
        Err(ValidateNameError::TooLong)
    } else {
        Ok(())
    }
}

/// Maximum length in bytes for an attribute value. Values may be empty.
pub const CLIENT_ATTRIBUTE_VALUE_MAX_LENGTH: usize = 128;
/// Maximum number of entries allowed in the attributes map.
pub const CLIENT_ATTRIBUTES_MAX_COUNT: usize = 128;

/// Error returned when an attributes map fails validation.
#[derive(Debug, thiserror::Error)]
pub enum ValidateAttributesError {
    #[error("Too many attributes (must be no more than {CLIENT_ATTRIBUTES_MAX_COUNT}).")]
    TooManyEntries,
    #[error("Invalid attribute key: {0}")]
    InvalidKey(#[from] ValidateNameError),
    #[error(
        "Attribute value too long (must be no more than {CLIENT_ATTRIBUTE_VALUE_MAX_LENGTH} bytes)."
    )]
    ValueTooLong,
}

fn validate_attributes(attrs: &BTreeMap<String, String>) -> Result<(), ValidateAttributesError> {
    if attrs.len() > CLIENT_ATTRIBUTES_MAX_COUNT {
        return Err(ValidateAttributesError::TooManyEntries);
    }
    for (k, v) in attrs {
        validate_name(k)?;
        if v.len() > CLIENT_ATTRIBUTE_VALUE_MAX_LENGTH {
            return Err(ValidateAttributesError::ValueTooLong);
        }
    }
    Ok(())
}

#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum Error {
    #[error("Invalid endpoint name: {0}")]
    InvalidName(#[from] ValidateNameError),
    #[error("Invalid endpoint group: {0}")]
    InvalidGroup(ValidateNameError),
    #[error("Invalid endpoint attributes: {0}")]
    InvalidAttributes(#[from] ValidateAttributesError),
    #[error("Remote error: {0}")]
    Remote(#[from] RemoteError),
    #[error("Connection error: {0}")]
    Connect(#[from] ConnectError),
    #[error("Rpc error: {0}")]
    Rpc(#[from] irpc::Error),
    #[error(transparent)]
    Other(#[from] anyhow::Error),
    #[error("Local client actor is stopped, cannot send requests")]
    ActorStopped,
}

impl From<tokio::sync::mpsc::error::SendError<ClientActorMessage>> for Error {
    fn from(_value: tokio::sync::mpsc::error::SendError<ClientActorMessage>) -> Self {
        Error::ActorStopped
    }
}

impl From<tokio::sync::oneshot::error::RecvError> for Error {
    fn from(_value: tokio::sync::oneshot::error::RecvError) -> Self {
        Error::ActorStopped
    }
}

impl Client {
    pub fn builder(endpoint: &Endpoint) -> ClientBuilder {
        ClientBuilder::new(endpoint)
    }

    /// Read the current endpoint name from the local client.
    pub async fn name(&self) -> Result<Option<String>, Error> {
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::ReadName { done: tx })
            .await?;
        rx.await.map_err(Into::into)
    }

    /// Read the current endpoint group from the local client.
    pub async fn group(&self) -> Result<Option<String>, Error> {
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::ReadGroup { done: tx })
            .await?;
        rx.await.map_err(Into::into)
    }

    /// Name the active endpoint cloud-side.
    ///
    /// names can be any UTF-8 string, with a min length of 2 bytes, and
    /// maximum length of 128 bytes. **name uniqueness is not enforced.**
    pub async fn set_name(&self, name: impl Into<String>) -> Result<(), Error> {
        let name = name.into();
        validate_name(&name)?;
        debug!(name_len = name.len(), "calling set name");
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::NameEndpoint { name, done: tx })
            .await?;
        rx.await?
    }

    /// Attach the active endpoint to a single named group cloud-side.
    ///
    /// A group name must be 2 to 128 bytes of UTF-8.
    pub async fn set_group(&self, group: impl Into<String>) -> Result<(), Error> {
        let group: String = group.into();
        validate_name(&group).map_err(Error::InvalidGroup)?;
        debug!(%group, "calling set group");
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::SetGroup { group, done: tx })
            .await?;
        rx.await?
    }

    /// Replace the arbitrary key-value attributes on the active endpoint cloud-side.
    ///
    /// Accepts any iterable of `(key, value)` pairs (arrays of tuples, `Vec`s,
    /// `HashMap`s, `BTreeMap`s, etc.), so most calls fit on a single line:
    ///
    /// ```no_run
    /// # use iroh_services::Client;
    /// # async fn example(client: Client) -> anyhow::Result<()> {
    /// client
    ///     .set_attributes([("env", "prod"), ("region", "us-west")])
    ///     .await?;
    /// # Ok(()) }
    /// ```
    ///
    /// Each key must be 2 to 128 bytes of UTF-8; values may be empty and are limited
    /// to 128 bytes; at most 128 entries are allowed. Each call fully replaces
    /// the prior set; passing an empty iterator clears all attributes.
    pub async fn set_attributes<I, K, V>(&self, attrs: I) -> Result<(), Error>
    where
        I: IntoIterator<Item = (K, V)>,
        K: Into<String>,
        V: Into<String>,
    {
        let collected: BTreeMap<String, String> = attrs
            .into_iter()
            .map(|(k, v)| (k.into(), v.into()))
            .collect();
        validate_attributes(&collected)?;
        debug!(attr_count = collected.len(), "calling set attributes");
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::SetAttributes {
                attributes: collected,
                done: tx,
            })
            .await?;
        rx.await?
    }

    /// Set or replace a single attribute, merging it into the endpoint's existing
    /// attributes rather than replacing the whole set.
    ///
    /// A convenience over [`set_attributes`](Self::set_attributes) when you only
    /// need to change one value. The key must be 2 to 128 bytes of UTF-8 and the
    /// value is limited to 128 bytes; the merged set must stay within 128 entries.
    pub async fn set_attribute(
        &self,
        key: impl Into<String>,
        value: impl Into<String>,
    ) -> Result<(), Error> {
        // Validation happens in the actor against the merged set (current
        // attributes plus this entry), since only there is the current set
        // known. Merging can exceed the entry-count limit even when this
        // single entry is valid.
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::SetAttribute {
                key: key.into(),
                value: value.into(),
                done: tx,
            })
            .await?;
        rx.await?
    }

    /// Pings the remote node.
    pub async fn ping(&self) -> Result<Pong, Error> {
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::Ping { done: tx })
            .await?;
        rx.await?
    }

    /// immediately send a single dump of metrics to iroh-services. It's not necessary
    /// to call this function if you're using a non-zero metrics interval,
    /// which will automatically propagate metrics on the set interval for you
    pub async fn push_metrics(&self) -> Result<(), Error> {
        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::SendMetrics { done: tx })
            .await?;
        rx.await?
    }

    /// Shuts down the client, after pushing one final round of metrics.
    ///
    /// Calling this aborts any inflight requests, and all subsequent requests
    /// triggered via methods on [`Self`] will fail. If the client is currently
    /// connected, it will send one final metrics update before shutting down
    /// the client actor.
    ///
    /// Dropping the client without calling this method immediately stops the
    /// actor without a final metrics push.
    pub async fn shutdown(&self) {
        self.shutdown.cancel();
        // The actor drops the inbox receiver on its way out.
        self.message_channel.closed().await;
    }

    /// Grant capabilities to a remote endpoint. Creates a signed RCAN token
    /// and sends it to iroh-services for storage. The remote can then use this token
    /// when dialing back to authorize its requests.
    pub async fn grant_capability(
        &self,
        remote_id: EndpointId,
        caps: impl IntoIterator<Item = impl Into<crate::caps::Cap>>,
    ) -> Result<(), Error> {
        let cap = crate::caps::create_grant_token(
            self.endpoint.secret_key().clone(),
            remote_id,
            DEFAULT_CAP_EXPIRY,
            Caps::new(caps),
        )
        .map_err(Error::Other)?;

        let (tx, rx) = oneshot::channel();
        self.message_channel
            .send(ClientActorMessage::GrantCap {
                cap: Box::new(cap),
                done: tx,
            })
            .await?;
        rx.await?
    }

    /// run local network status diagnostics, optionally uploading the results
    pub async fn net_diagnostics(&self, send: bool) -> Result<DiagnosticsReport, Error> {
        let report = run_diagnostics(&self.endpoint).await?;
        if send {
            let (tx, rx) = oneshot::channel();
            self.message_channel
                .send(ClientActorMessage::PutNetworkDiagnostics {
                    done: tx,
                    report: Box::new(report.clone()),
                })
                .await?;
            rx.await??;
        }

        Ok(report)
    }
}

enum ClientActorMessage {
    SendMetrics {
        done: oneshot::Sender<Result<(), Error>>,
    },
    Ping {
        done: oneshot::Sender<Result<Pong, Error>>,
    },
    GrantCap {
        // boxed to avoid large enum variants
        cap: Box<Rcan<Caps>>,
        done: oneshot::Sender<Result<(), Error>>,
    },
    PutNetworkDiagnostics {
        report: Box<DiagnosticsReport>,
        done: oneshot::Sender<Result<(), Error>>,
    },
    ReadName {
        done: oneshot::Sender<Option<String>>,
    },
    ReadGroup {
        done: oneshot::Sender<Option<String>>,
    },
    NameEndpoint {
        name: String,
        done: oneshot::Sender<Result<(), Error>>,
    },
    SetGroup {
        group: String,
        done: oneshot::Sender<Result<(), Error>>,
    },
    SetAttributes {
        attributes: BTreeMap<String, String>,
        done: oneshot::Sender<Result<(), Error>>,
    },
    SetAttribute {
        key: String,
        value: String,
        // Carries the full client `Error` (not just `RemoteError`) because the
        // merged-set validation happens in the actor, where the current set is
        // known, and can fail with a local `InvalidAttributes` error.
        done: oneshot::Sender<Result<(), Error>>,
    },
}

/// Whether an rpc error means the connection can no longer carry requests.
///
/// Most of these are the stream pair or the connection itself failing, so the
/// connection has to go. An oversized message is the exception: irpc rejects it
/// locally before anything reaches the wire, so the connection is still fine.
/// Re-dialing on it would pay for a handshake and a full metrics schema resend
/// only to hit the same limit on the retry.
fn is_connection_lost(err: &irpc::Error) -> bool {
    !matches!(
        err,
        irpc::Error::Send {
            source: irpc::channel::SendError::MaxMessageSizeExceeded { .. },
            ..
        }
    )
}

/// An irpc client bound to a single authenticated connection.
///
/// The server accepts requests only after an `Auth` as the very first
/// request on a connection, so [`RpcClient::connect`] dials and
/// authenticates in one step; a value of this type never exists
/// unauthenticated.
struct RpcClient {
    /// Kept alongside the irpc client to detect a remote close.
    connection: Connection,
    irpc: IrohServicesClient,
}

impl RpcClient {
    /// Dials the remote and authenticates as the connection's first request.
    async fn connect(
        endpoint: &Endpoint,
        remote: EndpointAddr,
        caps: Rcan<Caps>,
    ) -> Result<Self, Error> {
        trace!("client connecting and authorizing");
        let connection = endpoint
            .connect(remote, ALPN)
            .await
            .inspect_err(|err| debug!("connect failed: {err:?}"))?;
        let irpc = IrohServicesClient::boxed(IrohRemoteConnection::new(connection.clone()));
        irpc.rpc(Auth { caps })
            .await
            .inspect_err(|err| debug!("authorization failed: {err:?}"))
            .map_err(|err| RemoteError::AuthError(err.to_string()))?;
        Ok(Self { connection, irpc })
    }
}

struct ClientActor {
    capabilities: Rcan<Caps>,
    endpoint: Endpoint,
    remote: EndpointAddr,
    /// The active authenticated connection, established on demand.
    ///
    /// The actor owns the connection lifecycle rather than using a lazy
    /// reconnecting client: the server requires `Auth` as the first request
    /// on every connection, so a transparent mid-request reconnect would send
    /// an unauthenticated request that the server rejects. [`Self::connect`]
    /// establishes this; [`Self::rpc`] clears it on transport errors so the
    /// next request re-dials and re-authenticates.
    client: Option<RpcClient>,
    name: Option<String>,
    group: Option<String>,
    attributes: BTreeMap<String, String>,
    session_id: Uuid,
    encoder: Encoder,
    /// Kept so connect() can rebuild the encoder to re-send the metrics schema.
    registry: Arc<RwLock<Registry>>,
}

impl ClientActor {
    /// Runs the actor until the inbox closes or `shutdown` is cancelled, then
    /// pushes one final round of metrics.
    ///
    /// Cancelling `shutdown` drops whatever [`Self::run_inner`] is awaiting,
    /// including all in-flight or queued requests. We do this so that a
    /// pending dial does not hold up shutdown.
    async fn run(
        mut self,
        interval: Option<Duration>,
        mut inbox: tokio::sync::mpsc::Receiver<ClientActorMessage>,
        shutdown: CancellationToken,
    ) {
        let metrics_enabled = interval.is_some();
        let shutdown_and_grace_period_expired = async {
            shutdown.cancelled().await;
            time::sleep(SHUTDOWN_GRACE).await;
        };

        let clean_shutdown = tokio::select! {
            () = self.run_inner(interval, &mut inbox, &shutdown) => true,
            () = shutdown_and_grace_period_expired => {
                debug!("shutdown grace elapsed, dropping the request in flight");
                false
            }
        };

        // Flush only over a connection that already exists. Dialing here would
        // reintroduce the stall that cancelling just avoided, and an endpoint
        // that never connected has nothing the server could attribute.
        if clean_shutdown && metrics_enabled && self.is_connected() {
            match time::timeout(SHUTDOWN_FLUSH_TIMEOUT, self.send_metrics()).await {
                Ok(Ok(())) => trace!("pushed final metrics on shutdown"),
                Ok(Err(err)) => debug!(%err, "failed to push final metrics on shutdown"),
                Err(_) => debug!("final metrics push on shutdown timed out"),
            }
        }
        debug!("client actor shut down");
    }

    async fn run_inner(
        &mut self,
        interval: Option<Duration>,
        inbox: &mut tokio::sync::mpsc::Receiver<ClientActorMessage>,
        shutdown: &CancellationToken,
    ) {
        let mut metrics_timer = interval.map(|interval| time::interval(interval));
        trace!("starting client actor");

        // Send the initial metadata (set via the builder) once the actor starts.
        // These live on `self`; a send failure here is logged, not fatal.
        if let Some(name) = self.name.clone()
            && let Err(err) = self.send_name_endpoint(name).await
        {
            warn!(err = %err, "failed setting endpoint name on startup");
        }

        if let Some(group) = self.group.clone()
            && let Err(err) = self.send_set_group(group).await
        {
            warn!(err = %err, "failed setting endpoint group on startup");
        }

        if !self.attributes.is_empty()
            && let Err(err) = self.send_set_attributes(self.attributes.clone()).await
        {
            warn!(err = %err, "failed setting endpoint attributes on startup");
        }

        loop {
            trace!("client actor tick");
            tokio::select! {
                biased;
                // Shutdown is only observed between requests: a branch body below
                // runs to completion, so a request already in flight finishes first.
                () = shutdown.cancelled() => {
                    trace!("client actor observed shutdown between requests");
                    break;
                }
                Some(msg) = inbox.recv() => {
                    match msg {
                        ClientActorMessage::Ping { done } => {
                            let res = self.send_ping().await;
                            done.send(res).ok();
                        },
                        ClientActorMessage::SendMetrics { done } => {
                            trace!("sending metrics manually triggered");
                            let res = self.send_metrics().await;
                            done.send(res).ok();
                        }
                        ClientActorMessage::GrantCap { cap, done } => {
                            let res = self.grant_cap(*cap).await;
                            done.send(res).ok();
                        }
                        ClientActorMessage::ReadName { done } => {
                            done.send(self.name.clone()).ok();
                        }
                        ClientActorMessage::ReadGroup { done } => {
                            done.send(self.group.clone()).ok();
                        }
                        ClientActorMessage::NameEndpoint { name, done } => {
                            let res = self.send_name_endpoint(name).await;
                            done.send(res).ok();
                        }
                        ClientActorMessage::SetGroup { group, done } => {
                            let res = self.send_set_group(group).await;
                            done.send(res).ok();
                        }
                        ClientActorMessage::SetAttributes { attributes, done } => {
                            let res = self.send_set_attributes(attributes).await;
                            done.send(res).ok();
                        }
                        ClientActorMessage::SetAttribute { key, value, done } => {
                            // Merge into the current set and validate the union:
                            // adding one valid entry to a valid set can still
                            // exceed the max entry count, so the single entry
                            // being valid is not enough.
                            let mut merged = self.attributes.clone();
                            merged.insert(key, value);
                            let res = match validate_attributes(&merged) {
                                Ok(()) => self.send_set_attributes(merged).await,
                                Err(err) => Err(Error::from(err)),
                            };
                            done.send(res).ok();
                        }
                        ClientActorMessage::PutNetworkDiagnostics { report, done } => {
                            let res = self.put_network_diagnostics(*report).await;
                            done.send(res).ok();
                        }
                    }
                }
                _ = async {
                    if let Some(ref mut timer) = metrics_timer {
                        timer.tick().await;
                    } else {
                        std::future::pending::<()>().await;
                    }
                } => {
                    trace!("metrics send tick");
                    if let Err(err) = self.send_metrics().await {
                        debug!("failed to push metrics: {:#?}", err);
                    }
                },
            }
        }
    }

    /// Whether a connection is established and still usable.
    ///
    /// Distinct from holding a [`RpcClient`]: the remote may have closed the
    /// connection since, in which case sending over it would have to re-dial.
    fn is_connected(&self) -> bool {
        self.client
            .as_ref()
            .is_some_and(|client| client.connection.close_reason().is_none())
    }

    /// Returns the client for the active connection, establishing one if needed.
    ///
    /// The server keeps one metrics decoder per connection, so a fresh
    /// connection also gets a fresh encoder: the next metrics export then
    /// carries the full schema for the server's fresh decoder.
    ///
    /// # Errors
    ///
    /// Returns an error when the dial, or the authentication that follows it,
    /// fails. A connection is stored only once both have succeeded, and a
    /// connection found closed is cleared before either is attempted, so a
    /// caller that sees an error has nothing left to clean up.
    async fn connect(&mut self) -> Result<&IrohServicesClient, Error> {
        // A connection the remote has closed (server restart, error close)
        // can't carry requests anymore; drop it and re-dial.
        if let Some(client) = &self.client
            && let Some(reason) = client.connection.close_reason()
        {
            debug!(%reason, "connection closed by remote, reconnecting");
            self.client = None;
        }
        // Taking the connection out and putting it back lets `insert` hand out
        // a borrow tied to the value it just stored. Checking `is_none` and
        // then looking the value up again needs an `expect`, because the
        // compiler cannot see that the lookup follows the store.
        let client = match self.client.take() {
            Some(client) => client,
            None => {
                let client = RpcClient::connect(
                    &self.endpoint,
                    self.remote.clone(),
                    self.capabilities.clone(),
                )
                .await?;
                // Recreate the metrics encoder to force sending the schema on the next metrics push.
                self.encoder = Encoder::new(self.registry.clone());
                client
            }
        };
        Ok(&self.client.insert(client).irpc)
    }

    async fn rpc<Req, Res>(&mut self, msg: Req) -> Result<Res, Error>
    where
        IrohServicesProtocol: From<Req>,
        ServicesMessage: From<WithChannels<Req, IrohServicesProtocol>>,
        Req: Channels<
                IrohServicesProtocol,
                Tx = irpc::channel::oneshot::Sender<Res>,
                Rx = NoReceiver,
            > + Display,
        Res: RpcMessage,
    {
        trace!(request = %msg, "client actor send request");
        let client = self.connect().await?;
        let res = client.rpc(msg).await;

        if let Err(err) = &res
            && is_connection_lost(err)
        {
            // The connection is gone or in an unknown state. Clear the client
            // so that the next request dials and authenticates.
            self.client = None;
        }

        res.inspect_err(|err| warn!("rpc error: {err}"))
            .map_err(Error::from)
    }

    async fn send_ping(&mut self) -> Result<Pong, Error> {
        let req = rand::random();
        self.rpc(Ping { req_id: req }).await
    }

    async fn send_name_endpoint(&mut self, name: String) -> Result<(), Error> {
        self.rpc(NameEndpoint { name: name.clone() }).await??;
        self.name = Some(name);
        Ok(())
    }

    async fn send_set_group(&mut self, group: String) -> Result<(), Error> {
        self.rpc(SetGroup {
            group: group.clone(),
        })
        .await??;
        self.group = Some(group);
        Ok(())
    }

    async fn send_set_attributes(
        &mut self,
        attributes: BTreeMap<String, String>,
    ) -> Result<(), Error> {
        self.rpc(SetAttributes {
            attributes: attributes.clone(),
        })
        .await??;
        self.attributes = attributes;
        Ok(())
    }

    async fn send_metrics(&mut self) -> Result<(), Error> {
        // Connecting replaces the encoder, so it must happen before the
        // export: the first update on a fresh connection has to carry the
        // schema for the server's fresh decoder.
        self.connect().await?;
        let update = self.encoder.export();
        // let delta = update_delta(&self.latest_ackd_update, &update);
        let req = PutMetrics {
            session_id: self.session_id,
            update,
        };
        self.rpc(req).await??;
        Ok(())
    }

    async fn grant_cap(&mut self, cap: Rcan<Caps>) -> Result<(), Error> {
        self.rpc(GrantCap { cap }).await??;
        Ok(())
    }

    async fn put_network_diagnostics(&mut self, report: DiagnosticsReport) -> Result<(), Error> {
        self.rpc(PutNetworkDiagnostics { report }).await??;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::HashMap,
        sync::{
            Arc, RwLock,
            atomic::{AtomicBool, Ordering},
        },
    };

    use iroh::{
        Endpoint, EndpointAddr, SecretKey,
        endpoint::{Connection, presets},
        protocol::{AcceptError, ProtocolHandler, Router},
    };
    use iroh_metrics::{
        Registry,
        encoding::{Decoder, Encoder},
    };
    use irpc::WithChannels;
    use irpc_iroh::read_request;
    use n0_error::AnyError;
    use n0_future::{
        task,
        time::{self, Duration},
    };
    use rand::{RngExt, SeedableRng};
    use temp_env_vars::temp_env_vars;

    use crate::{
        Client, ClientBuilder,
        api_secret::ApiSecret,
        caps::{Cap, Caps, create_api_token_from_secret_key},
        client::{
            API_SECRET_ENV_VAR_NAME, BuildError, CLIENT_ATTRIBUTE_VALUE_MAX_LENGTH,
            CLIENT_ATTRIBUTES_MAX_COUNT, CLIENT_NAME_MAX_LENGTH, Error, ValidateAttributesError,
            ValidateNameError, is_connection_lost,
        },
        protocol::{ALPN, IrohServicesProtocol, Pong, ServicesMessage},
    };

    /// What the test server recorded about one PutMetrics request.
    #[derive(Debug)]
    struct SeenUpdate {
        has_schema: bool,
        decoded_items: usize,
    }

    /// What the test server recorded, in arrival order.
    #[derive(Debug)]
    enum Seen {
        Metrics(SeenUpdate),
        /// A Ping whose response reached the client without the stream failing.
        PingAnswered,
    }

    /// In-process stand-in for the services backend.
    ///
    /// Mirrors the real server's session rules: the first request on every
    /// connection must be Auth, a repeat Auth on a live connection closes it,
    /// and the metrics decoder lives per connection. Requests are read one at
    /// a time and any error ends the connection, as in the backend, so a
    /// client that drops a request mid-flight takes the connection with it.
    #[derive(Debug)]
    struct TestServer {
        seen: tokio::sync::mpsc::UnboundedSender<Seen>,
        /// Kills the next connection without answering, simulating a restart.
        drop_next: Arc<AtomicBool>,
        /// Held before answering a Ping, so a test can catch one in flight.
        ping_delay: Duration,
    }

    impl TestServer {
        fn new(seen: tokio::sync::mpsc::UnboundedSender<Seen>) -> Self {
            Self {
                seen,
                drop_next: Arc::new(AtomicBool::new(false)),
                ping_delay: Duration::ZERO,
            }
        }

        fn ping_delay(mut self, delay: Duration) -> Self {
            self.ping_delay = delay;
            self
        }

        async fn handle_connection(&self, connection: Connection) -> anyhow::Result<()> {
            let Some(first_request) = read_request::<IrohServicesProtocol>(&connection).await?
            else {
                return Ok(());
            };
            let ServicesMessage::Auth(WithChannels { tx, .. }) = first_request else {
                connection.close(400u32.into(), b"Expected initial auth message");
                return Ok(());
            };
            tx.send(()).await?;

            let mut decoder = Decoder::default();
            loop {
                let Ok(Some(request)) = read_request::<IrohServicesProtocol>(&connection).await
                else {
                    return Ok(());
                };
                if self.drop_next.swap(false, Ordering::SeqCst) {
                    // Simulates a server restart: the connection dies without an
                    // answer and takes the per-connection decoder with it.
                    connection.close(500u32.into(), b"test restart");
                    return Ok(());
                }
                match request {
                    ServicesMessage::Auth(_) => {
                        connection.close(400u32.into(), b"Unexpected auth message");
                        anyhow::bail!("client re-sent auth on a live connection");
                    }
                    ServicesMessage::Ping(WithChannels { inner, tx, .. }) => {
                        time::sleep(self.ping_delay).await;
                        // A client that dropped this request has reset the
                        // stream, and this send is where the server finds out.
                        tx.send(Pong {
                            req_id: inner.req_id,
                        })
                        .await?;
                        let _ = self.seen.send(Seen::PingAnswered);
                    }
                    ServicesMessage::PutMetrics(WithChannels { inner, tx, .. }) => {
                        let has_schema = inner.update.schema.is_some();
                        decoder.import(inner.update);
                        let _ = self.seen.send(Seen::Metrics(SeenUpdate {
                            has_schema,
                            decoded_items: decoder.iter().count(),
                        }));
                        tx.send(Ok(())).await?;
                    }
                    _ => {
                        connection.close(400u32.into(), b"Unexpected message in test");
                        anyhow::bail!("unexpected message in test");
                    }
                }
            }
        }
    }

    impl ProtocolHandler for TestServer {
        async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
            self.handle_connection(connection).await.map_err(|e| {
                let boxed: Box<dyn std::error::Error + Send + Sync> = e.into();
                AcceptError::from(AnyError::from(boxed))
            })
        }
    }

    /// Spawns `server` on its own endpoint and returns a client builder aimed
    /// at it, with the router and the client endpoint for teardown.
    ///
    /// The metrics interval is left to the caller, since that is what the
    /// tests vary. The test server accepts any capability, so a self-issued
    /// token works.
    async fn spawn_test_server(seed: u64, server: TestServer) -> (Router, Endpoint, ClientBuilder) {
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(seed);
        let server_ep = Endpoint::builder(presets::Minimal).bind().await.unwrap();
        let client_ep = Endpoint::builder(presets::Minimal).bind().await.unwrap();
        let router = Router::builder(server_ep.clone())
            .accept(ALPN, server)
            .spawn();

        let shared_secret = SecretKey::from_bytes(&rng.random());
        let cap = create_api_token_from_secret_key(
            shared_secret,
            client_ep.id(),
            Duration::from_secs(3600),
            Caps::for_shared_secret(),
        )
        .unwrap();

        let builder = Client::builder(&client_ep)
            .remote(server_ep.addr())
            .rcan(cap)
            .unwrap();
        (router, client_ep, builder)
    }

    /// Awaits the next metrics push the server recorded.
    async fn next_metrics(rx: &mut tokio::sync::mpsc::UnboundedReceiver<Seen>) -> SeenUpdate {
        match rx.recv().await.expect("server dropped the record channel") {
            Seen::Metrics(update) => update,
            other => panic!("expected a metrics push, recorded {other:?}"),
        }
    }

    /// Takes everything the server has recorded so far, without waiting.
    fn recorded_so_far(rx: &mut tokio::sync::mpsc::UnboundedReceiver<Seen>) -> Vec<Seen> {
        let mut seen = Vec::new();
        while let Ok(record) = rx.try_recv() {
            seen.push(record);
        }
        seen
    }

    /// A reconnect must make the next metrics update carry the schema.
    ///
    /// The server holds one decoder per connection, so the first update after
    /// a re-auth is undecodable unless the schema comes along again.
    #[tokio::test]
    async fn test_metrics_schema_resent_after_reconnect() {
        let (seen_tx, mut seen_rx) = tokio::sync::mpsc::unbounded_channel();
        let server = TestServer::new(seen_tx);
        let drop_next = server.drop_next.clone();
        let (router, client_ep, builder) = spawn_test_server(2, server).await;

        let client = builder.disable_metrics_interval().build().await.unwrap();

        // The first export on a fresh session carries the schema.
        client.push_metrics().await.unwrap();
        let first = next_metrics(&mut seen_rx).await;
        assert!(first.has_schema);
        assert!(first.decoded_items > 0);

        // Steady state stops sending the schema. The endpoint's own metric
        // families may still grow for a few exports right after connecting
        // (each growth re-publishes the schema), so push until the schema
        // settles; the connection's decoder keeps decoding throughout.
        let mut settled = false;
        for _ in 0..20 {
            client.push_metrics().await.unwrap();
            let seen = next_metrics(&mut seen_rx).await;
            assert!(seen.decoded_items > 0);
            if !seen.has_schema {
                settled = true;
                break;
            }
        }
        assert!(settled, "schema must stop being sent once it is unchanged");

        // The server drops the connection mid-request: one failed round trip.
        drop_next.store(true, Ordering::SeqCst);
        assert!(client.push_metrics().await.is_err());

        // The client re-dials and re-auths; the first update on the new
        // connection must include the schema for the server's fresh decoder.
        client.push_metrics().await.unwrap();
        let third = next_metrics(&mut seen_rx).await;
        assert!(third.has_schema, "schema must be re-sent after a reconnect");
        assert!(third.decoded_items > 0);

        router.shutdown().await.unwrap();
        client_ep.close().await;
    }

    /// Documents the encoder and decoder contract the reconnect fix relies on.
    #[tokio::test]
    async fn test_fresh_encoder_resends_schema() {
        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();
        let mut registry = Registry::default();
        registry.register_all(endpoint.metrics());
        let registry = Arc::new(RwLock::new(registry));

        let mut encoder = Encoder::new(registry.clone());
        let first = encoder.export();
        assert!(first.schema.is_some());

        // Steady state omits the schema once it has been exported.
        let schemaless = encoder.export();
        assert!(schemaless.schema.is_none());

        // A decoder that never saw the schema decodes such an update to
        // nothing: this is the server side of the reconnect bug.
        let mut fresh_decoder = Decoder::default();
        fresh_decoder.import(schemaless);
        assert_eq!(fresh_decoder.iter().count(), 0);

        // A new encoder over the same registry starts at schema version zero
        // and re-publishes the schema, which is what auth() does on re-auth.
        let mut encoder = Encoder::new(registry);
        let resent = encoder.export();
        assert!(resent.schema.is_some());
        let mut fresh_decoder = Decoder::default();
        fresh_decoder.import(resent);
        assert!(fresh_decoder.iter().count() > 0);
    }

    #[tokio::test]
    #[temp_env_vars]
    async fn test_api_key_from_env() {
        // construct
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(0);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);
        unsafe {
            std::env::set_var(API_SECRET_ENV_VAR_NAME, api_secret.to_string());
        };

        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        let builder = Client::builder(&endpoint).api_secret_from_env().unwrap();

        let fake_endpoint_addr: EndpointAddr = fake_endpoint_id.into();
        assert_eq!(builder.remote, Some(fake_endpoint_addr));

        // Compare capability fields individually to avoid flaky timestamp
        // mismatches between the builder's rcan and a freshly-created one.
        let cap = builder.cap.as_ref().expect("expected capability to be set");
        assert_eq!(cap.capability(), &Caps::new([Cap::Client]));
        assert_eq!(cap.audience(), &endpoint.id().as_verifying_key());
        assert_eq!(cap.issuer(), &shared_secret.public().as_verifying_key());
    }

    /// Assert that disabling metrics interval can manually send metrics without
    /// panicking. Metrics sending itself is expected to fail.
    #[tokio::test]
    async fn test_no_metrics_interval() {
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(1);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);

        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        let client = Client::builder(&endpoint)
            .disable_metrics_interval()
            .api_secret(api_secret)
            .unwrap()
            .build()
            .await
            .unwrap();

        let err = client.push_metrics().await;
        assert!(err.is_err());
    }

    /// Assert that shutdown returns even when the final metrics push fails,
    /// and that the actor is stopped afterwards.
    #[tokio::test]
    async fn test_shutdown_stops_actor() {
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(2);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);

        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        let client = Client::builder(&endpoint)
            .api_secret(api_secret)
            .unwrap()
            .build()
            .await
            .unwrap();

        client.shutdown().await;

        // the actor is gone, so requests fail
        let err = client.push_metrics().await;
        assert!(err.is_err());
    }

    /// An oversized message never reaches the wire, so it must not cost the
    /// connection: re-dialing would pay for a handshake and a full metrics
    /// schema resend and then hit the same limit again.
    #[test]
    fn test_oversized_message_keeps_the_connection() {
        let send_error = |source| irpc::Error::Send {
            source,
            meta: Default::default(),
        };

        assert!(!is_connection_lost(&send_error(
            irpc::channel::SendError::MaxMessageSizeExceeded {
                meta: Default::default(),
            }
        )));
        assert!(is_connection_lost(&send_error(
            irpc::channel::SendError::ReceiverClosed {
                meta: Default::default(),
            }
        )));
    }

    /// Assert that shutting down with a request in flight lets that request
    /// finish, so the final metrics push still reaches the server.
    ///
    /// Dropping the request instead would reset its stream, and the server
    /// treats that as the connection failing: it stops reading, so the push
    /// that shutdown exists for is never seen.
    #[tokio::test]
    async fn test_shutdown_drains_request_in_flight() {
        let (seen_tx, mut seen_rx) = tokio::sync::mpsc::unbounded_channel();
        let (router, client_ep, builder) = spawn_test_server(
            9,
            // long enough for shutdown to land while a ping is in flight
            TestServer::new(seen_tx).ping_delay(Duration::from_millis(300)),
        )
        .await;

        let client = builder
            // long enough that only the startup tick fires on its own
            .metrics_interval(Duration::from_secs(3600))
            .build()
            .await
            .unwrap();

        // the startup push is what establishes the connection
        next_metrics(&mut seen_rx).await;

        // put a ping in flight, then shut down while the server sits on it
        let pinging = client.clone();
        let ping = task::spawn(async move { pinging.ping().await });
        time::sleep(Duration::from_millis(100)).await;
        client.shutdown().await;

        let recorded = recorded_so_far(&mut seen_rx);
        assert!(
            recorded
                .iter()
                .any(|seen| matches!(seen, Seen::PingAnswered)),
            "the ping in flight was dropped, so the server saw its stream fail"
        );
        assert!(
            recorded.iter().any(|seen| matches!(seen, Seen::Metrics(_))),
            "the final metrics push did not reach the server"
        );

        let _ = ping.await;
        router.shutdown().await.unwrap();
        client_ep.close().await;
    }

    /// Assert that shutdown does not wait for a request in flight.
    ///
    /// The metrics interval fires as soon as the actor starts, so with an
    /// unreachable remote the actor is inside a dial that runs for about a
    /// minute. Shutting down has to cancel that dial rather than queue behind
    /// it.
    #[tokio::test]
    async fn test_shutdown_cancels_dial_in_flight() {
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(3);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret, fake_endpoint_id);

        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        let client = Client::builder(&endpoint)
            .api_secret(api_secret)
            .unwrap()
            // TEST-NET-1, routable but silently dropped, so the dial hangs
            // instead of failing fast the way an address-less remote would.
            .remote(
                EndpointAddr::new(fake_endpoint_id).with_ip_addr("192.0.2.1:1234".parse().unwrap()),
            )
            .build()
            .await
            .unwrap();

        // let the startup metrics tick get as far as the dial
        time::sleep(Duration::from_millis(200)).await;

        time::timeout(Duration::from_secs(5), client.shutdown())
            .await
            .expect("shutdown blocked on the dial in flight");
    }

    #[tokio::test]
    async fn test_name() {
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(0);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);

        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        let builder = Client::builder(&endpoint)
            .name("my-node 👋")
            .unwrap()
            .api_secret(api_secret)
            .unwrap();

        assert_eq!(builder.name, Some("my-node 👋".to_string()));

        let Err(err) = Client::builder(&endpoint).name("a") else {
            panic!("name should fail for strings under 2 bytes");
        };
        assert!(matches!(
            err.downcast_ref::<BuildError>(),
            Some(BuildError::InvalidName(ValidateNameError::TooShort))
        ));

        let too_long_name = "👋".repeat(129);
        let Err(err) = Client::builder(&endpoint).name(&too_long_name) else {
            panic!("name should fail for strings over 128 bytes");
        };
        assert!(matches!(
            err.downcast_ref::<BuildError>(),
            Some(BuildError::InvalidName(ValidateNameError::TooLong))
        ));
    }

    #[tokio::test]
    async fn test_group() {
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(0);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret.clone(), fake_endpoint_id);

        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        let builder = Client::builder(&endpoint)
            .group("staging")
            .unwrap()
            .api_secret(api_secret)
            .unwrap();

        assert_eq!(builder.group, Some("staging".to_string()));

        let Err(err) = Client::builder(&endpoint).group("a") else {
            panic!("group should fail for strings under 2 bytes");
        };
        assert!(matches!(
            err.downcast_ref::<BuildError>(),
            Some(BuildError::InvalidGroup(ValidateNameError::TooShort))
        ));

        let too_long_group = "👋".repeat(129);
        let Err(err) = Client::builder(&endpoint).group(&too_long_group) else {
            panic!("group should fail for strings over 128 bytes");
        };
        assert!(matches!(
            err.downcast_ref::<BuildError>(),
            Some(BuildError::InvalidGroup(ValidateNameError::TooLong))
        ));
    }

    #[tokio::test]
    async fn test_attributes() {
        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        // empty iterator is accepted (clears attributes server-side)
        let builder = Client::builder(&endpoint)
            .attributes(std::iter::empty::<(String, String)>())
            .unwrap();
        assert_eq!(builder.attributes.as_ref().map(|m| m.len()), Some(0));

        // array literal of `&str` tuples, for the one-liner ergonomics
        let builder = Client::builder(&endpoint)
            .attributes([("env", "prod"), ("region", "us-west")])
            .unwrap();
        let attrs = builder.attributes.as_ref().expect("attributes set");
        assert_eq!(attrs.get("env").map(String::as_str), Some("prod"));
        assert_eq!(attrs.get("region").map(String::as_str), Some("us-west"));

        // HashMap<String, String> also works
        let mut map: HashMap<String, String> = HashMap::new();
        map.insert("k1".into(), "v1".into());
        map.insert("k2".into(), "".into()); // empty value is allowed
        let builder = Client::builder(&endpoint).attributes(map).unwrap();
        let attrs = builder.attributes.as_ref().expect("attributes set");
        assert_eq!(attrs.get("k2").map(String::as_str), Some(""));

        // value over 128 bytes errors
        let too_long_value = "x".repeat(129);
        let Err(err) = Client::builder(&endpoint).attributes([("ok", too_long_value.as_str())])
        else {
            panic!("attributes should fail for value over 128 bytes");
        };
        assert!(matches!(
            err.downcast_ref::<BuildError>(),
            Some(BuildError::InvalidAttributes(
                ValidateAttributesError::ValueTooLong
            ))
        ));

        // key under 2 bytes errors
        let Err(err) = Client::builder(&endpoint).attributes([("a", "v")]) else {
            panic!("attributes should fail for key under 2 bytes");
        };
        assert!(matches!(
            err.downcast_ref::<BuildError>(),
            Some(BuildError::InvalidAttributes(
                ValidateAttributesError::InvalidKey(ValidateNameError::TooShort)
            ))
        ));

        // more than 128 entries errors
        let big: Vec<(String, String)> = (0..(CLIENT_ATTRIBUTES_MAX_COUNT + 1))
            .map(|i| (format!("key_{i:04}"), format!("val_{i}")))
            .collect();
        let Err(err) = Client::builder(&endpoint).attributes(big) else {
            panic!("attributes should fail for more than 128 entries");
        };
        assert!(matches!(
            err.downcast_ref::<BuildError>(),
            Some(BuildError::InvalidAttributes(
                ValidateAttributesError::TooManyEntries
            ))
        ));
    }

    /// Build a client with no reachable server, mirroring `test_no_metrics_interval`.
    /// The runtime setters validate input locally before any network call, so
    /// validation errors surface without a live server.
    async fn build_serverless_client(seed: u64) -> Client {
        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(seed);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret, fake_endpoint_id);

        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();

        Client::builder(&endpoint)
            .disable_metrics_interval()
            .api_secret(api_secret)
            .unwrap()
            .build()
            .await
            .unwrap()
    }

    /// Covers the runtime `Client::set_group` path the builder tests miss:
    /// validation runs locally and returns `Error::InvalidGroup` without a server.
    #[tokio::test]
    async fn test_set_group_runtime_validation() {
        let client = build_serverless_client(2).await;

        let err = client
            .set_group("a")
            .await
            .expect_err("too-short group should fail validation");
        assert!(matches!(
            err,
            Error::InvalidGroup(ValidateNameError::TooShort)
        ));

        let too_long = "x".repeat(CLIENT_NAME_MAX_LENGTH + 1);
        let err = client
            .set_group(too_long)
            .await
            .expect_err("too-long group should fail validation");
        assert!(matches!(
            err,
            Error::InvalidGroup(ValidateNameError::TooLong)
        ));
    }

    /// Covers the runtime `Client::set_attributes` path the builder tests miss:
    /// validation runs locally and returns `Error::InvalidAttributes` without a server.
    #[tokio::test]
    async fn test_set_attributes_runtime_validation() {
        let client = build_serverless_client(3).await;

        // key under 2 bytes
        let err = client
            .set_attributes([("a", "v")])
            .await
            .expect_err("too-short attribute key should fail validation");
        assert!(matches!(
            err,
            Error::InvalidAttributes(ValidateAttributesError::InvalidKey(
                ValidateNameError::TooShort
            ))
        ));

        // value over the max length
        let too_long_value = "x".repeat(CLIENT_ATTRIBUTE_VALUE_MAX_LENGTH + 1);
        let err = client
            .set_attributes([("ok", too_long_value.as_str())])
            .await
            .expect_err("too-long attribute value should fail validation");
        assert!(matches!(
            err,
            Error::InvalidAttributes(ValidateAttributesError::ValueTooLong)
        ));

        // more entries than allowed
        let big: Vec<(String, String)> = (0..(CLIENT_ATTRIBUTES_MAX_COUNT + 1))
            .map(|i| (format!("key_{i:04}"), format!("val_{i}")))
            .collect();
        let err = client
            .set_attributes(big)
            .await
            .expect_err("too many attributes should fail validation");
        assert!(matches!(
            err,
            Error::InvalidAttributes(ValidateAttributesError::TooManyEntries)
        ));
    }

    #[tokio::test]
    async fn test_set_attribute_runtime_validation() {
        let client = build_serverless_client(7).await;

        // A bad single key is rejected before any network call.
        let err = client
            .set_attribute("a", "v")
            .await
            .expect_err("too-short attribute key should fail validation");
        assert!(matches!(
            err,
            Error::InvalidAttributes(ValidateAttributesError::InvalidKey(
                ValidateNameError::TooShort
            ))
        ));

        // A valid single attribute passes validation, then reaches the dial
        // (no server) and surfaces a connect error, proving set_attribute
        // is wired through the actor/RPC path.
        let err = client
            .set_attribute("firmware", "2.1.0")
            .await
            .expect_err("no server: remote call must fail after validation passes");
        assert!(matches!(err, Error::Connect(_)), "got {err:?}");
    }

    #[tokio::test]
    async fn test_set_attribute_merge_over_limit_rejected() {
        // A client already holding the maximum number of attributes.
        let full: Vec<(String, String)> = (0..CLIENT_ATTRIBUTES_MAX_COUNT)
            .map(|i| (format!("key_{i:04}"), "v".to_string()))
            .collect();

        let mut rng = rand::rngs::ChaCha8Rng::seed_from_u64(9);
        let shared_secret = SecretKey::from_bytes(&rng.random());
        let fake_endpoint_id = SecretKey::from_bytes(&rng.random()).public();
        let api_secret = ApiSecret::new(shared_secret, fake_endpoint_id);
        let endpoint = Endpoint::builder(presets::Minimal).bind().await.unwrap();
        let client = Client::builder(&endpoint)
            .disable_metrics_interval()
            .attributes(full)
            .unwrap()
            .api_secret(api_secret)
            .unwrap()
            .build()
            .await
            .unwrap();

        // Merging one more (individually valid) entry pushes the set over the
        // limit. The single-entry check would miss this; the merged-set check in
        // the actor catches it locally, before any network call.
        let err = client
            .set_attribute("one-too-many", "v")
            .await
            .expect_err("merging past the attribute limit must fail");
        assert!(
            matches!(
                err,
                Error::InvalidAttributes(ValidateAttributesError::TooManyEntries)
            ),
            "expected TooManyEntries, got {err:?}"
        );
    }

    /// Boundary "accepted" case for the runtime setter. Without a live server we
    /// cannot assert success; instead we assert the input passes local validation
    /// and the call proceeds to the (failing) dial, surfacing `Error::Connect`
    /// rather than an `Error::InvalidAttributes` validation error.
    #[tokio::test]
    async fn test_set_attributes_runtime_boundary_accepted() {
        let client = build_serverless_client(4).await;

        // value of exactly the max length is accepted by validation
        let max_value = "x".repeat(CLIENT_ATTRIBUTE_VALUE_MAX_LENGTH);
        let err = client
            .set_attributes([("ok".to_string(), max_value)])
            .await
            .expect_err("no server: remote call must fail after validation passes");
        assert!(
            matches!(err, Error::Connect(_)),
            "expected a connect error (validation accepted), got {err:?}"
        );

        // exactly CLIENT_ATTRIBUTES_MAX_COUNT entries is accepted by validation
        let max_entries: Vec<(String, String)> = (0..CLIENT_ATTRIBUTES_MAX_COUNT)
            .map(|i| (format!("key_{i:04}"), format!("val_{i}")))
            .collect();
        let err = client
            .set_attributes(max_entries)
            .await
            .expect_err("no server: remote call must fail after validation passes");
        assert!(
            matches!(err, Error::Connect(_)),
            "expected a connect error (validation accepted), got {err:?}"
        );
    }
}