heddle_thread_api/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Native v2 client design. Endpoint ownership, credentials and connection
3//! discovery belong to the application; this crate never obtains a Weft token.
4#[cfg(feature = "native")]
5pub mod authority;
6#[cfg(feature = "replication")]
7pub mod authority_admission;
8#[cfg(feature = "semantic-analysis")]
9pub mod behavior;
10#[cfg(feature = "replication")]
11pub mod boundary_acceptance;
12#[cfg(feature = "replication")]
13pub mod collaboration;
14pub mod content;
15#[cfg(feature = "replication")]
16pub mod creation;
17#[cfg(feature = "signing")]
18pub mod credentials;
19#[cfg(feature = "replication")]
20pub mod evidence;
21#[cfg(feature = "source-transfer")]
22pub mod fetch;
23pub mod hybrid;
24#[cfg(feature = "replication")]
25pub mod live_replication;
26pub mod observation;
27#[cfg(feature = "root-attachment")]
28pub mod pairing;
29#[cfg(any(feature = "native", feature = "replication", feature = "iroh"))]
30pub mod publication;
31mod reopen;
32#[cfg(feature = "replication")]
33pub mod replication;
34#[cfg(all(feature = "native", feature = "iroh"))]
35pub mod replication_rpc;
36#[cfg(feature = "signing")]
37pub mod request_proof;
38#[cfg(feature = "root-attachment")]
39pub mod root_attachment;
40#[cfg(feature = "replication")]
41pub mod thread_control;
42#[cfg(feature = "replication")]
43pub mod thread_ownership;
44pub mod transport;
45
46use api::v2::client::{Client, ClientError, RpcTransport};
47pub use api::{
48    heddle::api::v1alpha2 as contract,
49    v2::{client::Rpc, rpc},
50};
51use contract::{DescribeEndpointRequest, DescribeEndpointResponse, EndpointKind, ThreadRef};
52pub use reopen::is_reopen_retryable;
53use transport::Error;
54
55/// One authenticated source. A combined Thread retains one source per endpoint;
56/// a hosted action is sent directly to the Weft source, never via a device.
57pub struct Remote<T: RpcTransport<Error = Error>> {
58    pub api: Client<T>,
59    pub description: DescribeEndpointResponse,
60}
61
62impl<T: RpcTransport<Error = Error>> Remote<T> {
63    /// The caller supplies the Iroh-authenticated endpoint key and intended kind.
64    /// Describe is the only bootstrap exception to implemented-method discovery.
65    pub async fn discover(
66        transport: T,
67        endpoint_key: [u8; 32],
68        kind: EndpointKind,
69    ) -> Result<Self, ClientError<Error>> {
70        let bytes = transport
71            .unary(
72                rpc::EndpointServiceDescribeEndpoint::METHOD,
73                prost::Message::encode_to_vec(&DescribeEndpointRequest {
74                    understood_packages: vec!["heddle.api.v1alpha2".into()],
75                }),
76            )
77            .await
78            .map_err(ClientError::Transport)?;
79        let description: DescribeEndpointResponse = prost::Message::decode(bytes.as_slice())?;
80        Self::from_description(transport, endpoint_key, kind, description)
81    }
82
83    /// Reuse an authenticated description retained by the application. The
84    /// application owns its validity window; identity and package checks apply
85    /// exactly as they do to a fresh DescribeEndpoint response.
86    pub fn from_description(
87        transport: T,
88        endpoint_key: [u8; 32],
89        kind: EndpointKind,
90        description: DescribeEndpointResponse,
91    ) -> Result<Self, ClientError<Error>> {
92        if description
93            .endpoint
94            .as_ref()
95            .is_none_or(|source| source.public_key != endpoint_key || source.kind != kind as i32)
96            || !description
97                .supported_packages
98                .iter()
99                .any(|p| p == "heddle.api.v1alpha2")
100        {
101            return Err(ClientError::Transport(Error::Protocol(
102                "endpoint identity/package mismatch",
103            )));
104        }
105        let mut api = Client::new(transport, description.implemented_methods.clone());
106        if let Some(protocol) = &description.protocol {
107            api = api.with_protocol(protocol.clone());
108        }
109        Ok(Self { api, description })
110    }
111
112    /// Observe any contract view as atomic bounded batches. The bookmark binds
113    /// the exact method and projection as well as the authenticated endpoint.
114    pub async fn observe<M>(
115        &self,
116        mut request: M::Request,
117        resume: Option<observation::Resume>,
118    ) -> Result<observation::Observation<T::Reader, M::Response>, observation::Error>
119    where
120        M: api::v2::client::ServerStreamingRpc,
121        M::Request: observation::ObservationRequest,
122        M::Response: observation::ObservedEvent,
123    {
124        use observation::ObservationRequest as _;
125
126        use crate::reopen::ReopenRetryable as _;
127        let budget = observation::budget(&self.description)?;
128        let options = request.options_mut();
129        options.budget = Some(budget);
130        options.after_cursor.clear();
131        let mut query = b"heddle-observation-query-v2\0".to_vec();
132        query.extend_from_slice(M::METHOD.path.as_bytes());
133        query.push(0);
134        query.extend_from_slice(&prost::Message::encode_to_vec(&request));
135        observation::validate_resume(&resume, &self.description, &query)?;
136        if let Some(resume) = &resume {
137            request.options_mut().after_cursor = resume.cursor.clone();
138        }
139        let mut attempt = 0;
140        loop {
141            let messages = match self.api.observe::<M>(&request).await {
142                Ok(messages) => messages,
143                Err(error)
144                    if reopen::client_error_is_reopen_retryable(&error)
145                        && attempt + 1 < reopen::ATTEMPTS =>
146                {
147                    attempt += 1;
148                    reopen::backoff(attempt).await;
149                    continue;
150                }
151                Err(error) => return Err(error.into()),
152            };
153            let mut observation = observation::Observation::new(
154                messages,
155                &self.description,
156                budget,
157                resume.clone(),
158                query.clone(),
159            )?;
160            match observation.consume_open().await {
161                Ok(()) => return Ok(observation),
162                Err(error) if error.is_reopen_retryable() && attempt + 1 < reopen::ATTEMPTS => {
163                    attempt += 1;
164                    reopen::backoff(attempt).await;
165                }
166                Err(error) => {
167                    observation.prime_error(error);
168                    return Ok(observation);
169                }
170            }
171        }
172    }
173
174    /// Source-backed analysis uses the same committed view protocol as identity,
175    /// collaboration, checkouts and Thread observations.
176    pub async fn observe_analysis(
177        &self,
178        request: contract::ObserveAnalysisRequest,
179        resume: Option<observation::Resume>,
180    ) -> Result<observation::AnalysisObservation<T::Reader>, observation::Error> {
181        self.observe::<rpc::AnalysisServiceObserveAnalysis>(request, resume)
182            .await
183    }
184
185    /// Binding a Thread is local and costs no RPC. Persist this stable reference
186    /// when discovering/creating the Thread; its display name is never its key.
187    pub fn thread(&self, thread: ThreadRef) -> Thread<'_, T> {
188        Thread {
189            remote: self,
190            reference: thread,
191        }
192    }
193}
194
195pub struct Thread<'a, T: RpcTransport<Error = Error>> {
196    remote: &'a Remote<T>,
197    pub reference: ThreadRef,
198}
199
200impl<T: RpcTransport<Error = Error>> Thread<'_, T> {
201    /// Send a locally prepared original signed intent. Preparation consumes the
202    /// existing overview's field frontier and performs no extra RPC.
203    #[cfg(feature = "replication")]
204    pub async fn revise_intent(
205        &self,
206        command: &thread_control::PreparedControl,
207    ) -> Result<contract::ThreadMutationResponse, ClientError<Error>> {
208        let request = command.revise_intent().map_err(ClientError::Transport)?;
209        if request.thread.as_ref() != Some(&self.reference) {
210            return Err(ClientError::Transport(Error::Protocol(
211                "prepared command belongs to another Thread",
212            )));
213        }
214        self.remote
215            .api
216            .call::<rpc::ThreadServiceReviseIntent>(&request)
217            .await
218    }
219
220    pub async fn observe(
221        &self,
222        sections: &[contract::ThreadSection],
223        mode: contract::ObservationMode,
224        resume: Option<observation::Resume>,
225    ) -> Result<observation::ThreadObservation<T::Reader>, observation::Error> {
226        self.remote
227            .observe::<rpc::ThreadServiceObserveThread>(
228                contract::ObserveThreadRequest {
229                    thread: Some(self.reference.clone()),
230                    sections: sections.iter().map(|section| *section as i32).collect(),
231                    observe: Some(contract::ObserveOptions {
232                        mode: mode as i32,
233                        ..Default::default()
234                    }),
235                    ..Default::default()
236                },
237                resume,
238            )
239            .await
240    }
241}
heddle_thread_api/lib.rs

heddle_thread_api/
lib.rs
