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

weft_client/
hosted.rs
