weft_client/
hosted.rs

1// SPDX-License-Identifier: Apache-2.0
2//! One discovered v2 endpoint, with independent RPC streams on its connection.
3use std::time::Duration;
4
5use anyhow::{Context, Result, bail};
6use iroh::{
7    Endpoint, EndpointAddr, RelayMode,
8    endpoint::{Connection, presets},
9};
10use thread_api::{
11    Remote, contract::EndpointKind, credentials::Credentials, transport::IrohTransport,
12};
13
14use crate::{ConnectionOptions, descriptor_url, fetch_endpoint_descriptor};
15
16const ENDPOINT_CLOSE_TIMEOUT: Duration = Duration::from_secs(3);
17
18async fn close_endpoint(endpoint: &Endpoint) {
19    let _ = tokio::time::timeout(ENDPOINT_CLOSE_TIMEOUT, endpoint.close()).await;
20}
21
22/// A direct Weft connection. The application retains operation IDs, credentials,
23/// observed versions and stream checkpoints; reconnect never retries a mutation.
24pub struct HostedClient {
25    pub remote: Remote<IrohTransport<Credentials>>,
26    endpoint: Endpoint,
27    connection: Connection,
28}
29
30impl HostedClient {
31    /// Verify the application's HTTPS descriptor trust, connect once, and
32    /// discover the endpoint's actual v2 implementation inventory.
33    pub async fn connect_server(server: &str, options: &ConnectionOptions) -> Result<Self> {
34        let descriptor = fetch_endpoint_descriptor(
35            &descriptor_url(server)?,
36            &options.trusted_descriptors,
37            &options.http_client,
38        )
39        .await?;
40        Self::connect_endpoint(
41            descriptor.endpoint_addr()?,
42            options.credential.clone(),
43            options.timeout,
44            options.tls_ca_certificate_pem.as_deref(),
45        )
46        .await
47    }
48
49    /// Connect to an endpoint whose public key the application already trusts.
50    /// Relay addresses are routing hints; Iroh authenticates the supplied key.
51    pub async fn connect_endpoint(
52        address: EndpointAddr,
53        credential: Credentials,
54        timeout: Duration,
55        tls_ca_certificate_pem: Option<&str>,
56    ) -> Result<Self> {
57        if timeout.is_zero() {
58            bail!("hosted progress timeout must be positive");
59        }
60        let relays: Vec<_> = address.relay_urls().cloned().collect();
61        let relay_mode = if relays.is_empty() {
62            RelayMode::Disabled
63        } else {
64            RelayMode::custom(relays)
65        };
66        heddle_perf_contract::record_network_client_initialization();
67        let mut builder = Endpoint::builder(presets::Minimal).relay_mode(relay_mode);
68        if let Some(pem) = tls_ca_certificate_pem {
69            builder = builder.ca_tls_config(crate::relay_tls::ca_tls_config_from_pem(pem)?);
70        }
71        let endpoint = builder
72            .bind()
73            .await
74            .context("bind hosted client endpoint")?;
75        let connection =
76            match tokio::time::timeout(timeout, endpoint.connect(address, api::HOSTED_ALPN_V1))
77                .await
78                .context("hosted connection deadline elapsed")
79                .and_then(|connection| connection.context("connect to hosted endpoint"))
80            {
81                Ok(connection) => connection,
82                Err(error) => {
83                    close_endpoint(&endpoint).await;
84                    return Err(error);
85                }
86            };
87        let result = async {
88            let transport = IrohTransport::new(
89                connection.clone(),
90                credential,
91                api::framing::MAX_CONTROL_BODY,
92                timeout,
93            )?;
94            Remote::discover(
95                transport,
96                *connection.remote_id().as_bytes(),
97                EndpointKind::Weft,
98            )
99            .await
100            .context("discover hosted v2 API")
101        }
102        .await;
103        match result {
104            Ok(remote) => Ok(Self {
105                remote,
106                endpoint,
107                connection,
108            }),
109            Err(error) => {
110                connection.close(0u32.into(), b"v2 discovery failed");
111                close_endpoint(&endpoint).await;
112                Err(error)
113            }
114        }
115    }
116
117    pub fn local_endpoint_id(&self) -> iroh::EndpointId {
118        self.endpoint.id()
119    }
120
121    pub fn endpoint(&self) -> &Endpoint {
122        &self.endpoint
123    }
124
125    pub fn connection(&self) -> &Connection {
126        &self.connection
127    }
128
129    /// Release the discovered remote and the live Iroh session without closing.
130    ///
131    /// The caller becomes responsible for closing `connection` / `endpoint`.
132    pub fn into_parts(self) -> (Remote<IrohTransport<Credentials>>, Endpoint, Connection) {
133        let this = std::mem::ManuallyDrop::new(self);
134        // SAFETY: `this` is forgotten, so Drop will not close the connection.
135        // Each field is read exactly once.
136        unsafe {
137            (
138                std::ptr::read(&this.remote),
139                std::ptr::read(&this.endpoint),
140                std::ptr::read(&this.connection),
141            )
142        }
143    }
144
145    /// Cancel this connection's streams and await endpoint shutdown.
146    pub async fn close(&self) {
147        self.connection.close(0u32.into(), b"hosted client closed");
148        close_endpoint(&self.endpoint).await;
149    }
150}
151
152impl Drop for HostedClient {
153    fn drop(&mut self) {
154        self.connection.close(0u32.into(), b"hosted client dropped");
155    }
156}
weft_client/hosted.rs

weft_client/
hosted.rs
