heddle_thread_api/
replication_rpc.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Known-Thread replication on a directly authenticated Iroh connection.
3//! The host supplies its resolved owner/account authority and admission facets.
4//! Unknown Thread creation and source-object transfer have separate boundaries.
5use std::{collections::BTreeSet, future::Future, path::PathBuf, sync::Arc, time::Duration};
6
7use api::{
8    framing,
9    v2::client::{MessageReader, MessageWriter, RpcTransport},
10};
11use iroh::endpoint::Connection;
12use objects::{
13    object::thread_replication::{OPERATION_FORMAT, ThreadFacet},
14    store::ObjectStore,
15};
16use prost::Message;
17use repo::thread_replication::ThreadReplica;
18
19use crate::{
20    Rpc,
21    authority::RootAuthority,
22    contract::*,
23    live_replication::{self, Feed, Side},
24    replication::{
25        self, Session,
26        native::LocalReplica,
27        opening::{self, FRAME_LIMIT, validate_endpoint},
28    },
29    rpc,
30    transport::{self, Authorize, IrohTransport, Reader, Writer},
31};
32
33const TIMEOUT: Duration = Duration::from_secs(10);
34
35/// A resolved local Thread, with the admission scope chosen by its host.
36/// Export remains subject to the Thread's current per-destination sharing policy.
37#[derive(Clone)]
38pub struct Peer {
39    replica: ThreadReplica,
40    endpoint: EndpointRef,
41    facets: BTreeSet<ThreadFacet>,
42    genesis: ThreadGenesisRecord,
43    authority_home: Option<PathBuf>,
44}
45impl Peer {
46    pub fn new(
47        replica: ThreadReplica,
48        endpoint: EndpointRef,
49        facets: BTreeSet<ThreadFacet>,
50    ) -> Result<Self, transport::Error> {
51        validate_endpoint(&endpoint)?;
52        if facets.is_empty() {
53            return Err(transport::Error::Protocol(
54                "replication requires an admission facet",
55            ));
56        }
57        let genesis = replica.genesis_record().map_err(io_error)?;
58        Ok(Self {
59            replica,
60            endpoint,
61            facets,
62            genesis,
63            authority_home: None,
64        })
65    }
66
67    /// Enable original metadata author admission against this account's local
68    /// independently enrolled pin; never infer account ownership from delivery.
69    pub fn with_device_authority(mut self, home: PathBuf) -> Self {
70        self.authority_home = Some(home);
71        self
72    }
73    fn local_replica<S: ObjectStore + Send + Sync + 'static>(
74        &self,
75        store: Arc<S>,
76    ) -> LocalReplica<S> {
77        let local = LocalReplica::new(self.replica.clone(), store);
78        match &self.authority_home {
79            Some(home) => local.with_device_authority(home.clone()),
80            None => local,
81        }
82    }
83    fn reference(&self) -> Result<ThreadRef, transport::Error> {
84        let genesis = self.replica.genesis().map_err(store_error)?;
85        Ok(ThreadRef {
86            spool: Some(SpoolRef { id: genesis.spool }),
87            id: Some(ThreadId {
88                value: self.replica.thread_id().as_bytes().to_vec(),
89            }),
90        })
91    }
92
93    /// Drives one RPC until cancellation, transport failure, or live authority
94    /// revocation. Reopen with the same replica to repair from durable frontiers.
95    pub async fn connect<S, A, G, F>(
96        &self,
97        connection: Connection,
98        destination: EndpointKind,
99        signer: A,
100        store: Arc<S>,
101        feed: &Feed,
102        authorize: G,
103    ) -> live_replication::Result<(), replication::native::Error>
104    where
105        S: ObjectStore + Send + Sync + 'static,
106        A: Authorize,
107        G: Fn() -> F + Clone + Send + Sync + 'static,
108        F: Future<Output = Result<(), transport::Error>> + Send,
109    {
110        let remote_key = *connection.remote_id().as_bytes();
111        let destination = EndpointRef {
112            public_key: remote_key.to_vec(),
113            kind: destination as i32,
114        };
115        validate_endpoint(&destination)?;
116        let thread = self.reference()?;
117        let (_, version) = self.replica.sharing(&remote_key).map_err(store_error)?;
118        let opening = ReplicationOpen {
119            native_authority: None,
120            thread: Some(thread.clone()),
121            facets: self
122                .facets
123                .iter()
124                .copied()
125                .map(replication::wire_facet)
126                .collect(),
127            sharing_policy_version: version.map(|v| v.as_bytes().to_vec()).unwrap_or_default(),
128            thread_genesis: Some(self.genesis.clone()),
129            budget: Some(ReadBudget {
130                max_items: 64,
131                max_frame_bytes: FRAME_LIMIT as u32,
132                max_snapshot_bytes: 0,
133            }),
134            record_formats: vec![OPERATION_FORMAT.into()],
135            session_nonce: uuid::Uuid::new_v4().as_bytes().to_vec(),
136            source: Some(self.endpoint.clone()),
137            destination: Some(destination.clone()),
138            protocol: crate::hybrid::sync_protocol(),
139            import_authority: None,
140        };
141        let transport = IrohTransport::new(connection, signer, FRAME_LIMIT, TIMEOUT)?;
142        let requested_protocol = opening.protocol.clone();
143        let (writer, mut reader) = transport
144            .exchange(
145                rpc::SyncServiceReplicateThread::METHOD,
146                ReplicateThreadRequest {
147                    body: Some(replicate_thread_request::Body::Open(opening)),
148                }
149                .encode_to_vec(),
150            )
151            .await?;
152        let bytes = reader.next().await?.ok_or(transport::Error::Protocol(
153            "replication closed before Ready",
154        ))?;
155        let response =
156            ReplicateThreadResponse::decode(bytes.as_slice()).map_err(transport::Error::from)?;
157        let Some(replicate_thread_response::Body::Ready(ready)) = response.body else {
158            return Err(transport::Error::Protocol("replication requires Ready").into());
159        };
160        let (facets, max_items) =
161            opening::validate_ready(&ready, &thread, &destination, &self.facets, 64)?;
162        let session = Session::new(self.local_replica(store), remote_key, facets, max_items)?
163            .with_protocol(requested_protocol.as_ref(), ready.protocol.as_ref())?;
164        live_replication::run(
165            session,
166            reader,
167            writer,
168            Side::Initiator,
169            feed,
170            move |activity| {
171                let authorize = authorize.clone();
172                async move {
173                    if matches!(activity, live_replication::Activity::InputConsumed { .. }) {
174                        Ok(())
175                    } else {
176                        authorize().await
177                    }
178                }
179            },
180        )
181        .await
182    }
183
184    /// Accept exactly one ReplicateThread RPC. The connection must be routed
185    /// here by the host; other service methods are not advertised or emulated.
186    pub async fn accept<S>(
187        &self,
188        connection: Connection,
189        authority: Arc<RootAuthority>,
190        store: Arc<S>,
191        feed: &Feed,
192    ) -> live_replication::Result<(), replication::native::Error>
193    where
194        S: ObjectStore + Send + Sync + 'static,
195    {
196        let remote_key = *connection.remote_id().as_bytes();
197        let (send, mut recv) = tokio::time::timeout(TIMEOUT, connection.accept_bi())
198            .await
199            .map_err(|_| transport::Error::Timeout)?
200            .map_err(io_error)?;
201        let mut writer = Writer::new(send, FRAME_LIMIT, TIMEOUT);
202        let opening = async {
203            let mut prelude = vec![0; 6];
204            recv.read_exact(&mut prelude).await.map_err(io_error)?;
205            let method_size = u16::from_be_bytes([prelude[0], prelude[1]]) as usize;
206            let context_size =
207                u32::from_be_bytes([prelude[2], prelude[3], prelude[4], prelude[5]]) as usize;
208            if method_size == 0
209                || method_size > framing::MAX_METHOD_PATH
210                || context_size > framing::MAX_CALL_CONTEXT
211            {
212                return Err(transport::Error::Protocol(
213                    "opening exceeds metadata limits",
214                ));
215            }
216            prelude.resize(6 + method_size + context_size, 0);
217            recv.read_exact(&mut prelude[6..]).await.map_err(io_error)?;
218            let (request, _) = framing::decode_request_prelude(&prelude)?
219                .ok_or(transport::Error::Protocol("incomplete opening metadata"))?;
220            if request.method != rpc::SyncServiceReplicateThread::METHOD.path {
221                return Err(transport::Error::Protocol(
222                    "unsupported RPC on replication boundary",
223                ));
224            }
225            Ok(request.context)
226        };
227        let context = tokio::time::timeout(TIMEOUT, opening)
228            .await
229            .map_err(|_| transport::Error::Timeout)??;
230        let mut reader = Reader::for_method(
231            recv,
232            FRAME_LIMIT,
233            TIMEOUT,
234            rpc::SyncServiceReplicateThread::METHOD,
235        );
236        let bytes = reader
237            .next()
238            .await?
239            .ok_or(transport::Error::Protocol("missing replication opening"))?;
240        let peer = self.clone();
241        let verifier = authority.clone();
242        let checked = tokio::task::spawn_blocking(move || -> Result<_, transport::Error> {
243            let request = ReplicateThreadRequest::decode(bytes.as_slice())?;
244            let Some(replicate_thread_request::Body::Open(open)) = request.body else {
245                return Err(transport::Error::Protocol("replication requires Open"));
246            };
247            let (_, version) = peer.replica.sharing(&remote_key).map_err(store_error)?;
248            let accepted = opening::accept(
249                &open,
250                &peer.reference()?,
251                &peer.endpoint,
252                remote_key,
253                &peer.facets,
254                version.map(|v| v.as_bytes().to_vec()).unwrap_or_default(),
255            )?;
256            if accepted.import_authority.is_some() {
257                return Err(transport::Error::Protocol(
258                    "HYBRID replication requires independently selected hosted trust",
259                ));
260            }
261            let ready = accepted.ready;
262            let (facets, max_items) = opening::validate_ready(
263                &ready,
264                &peer.reference()?,
265                &peer.endpoint,
266                &peer.facets,
267                64,
268            )?;
269            let verified = verifier.verify(
270                &context,
271                rpc::SyncServiceReplicateThread::METHOD,
272                &bytes,
273                "write",
274                &peer.replica,
275            )?;
276            Ok((verified, facets, max_items, ready))
277        })
278        .await
279        .map_err(|e| transport::Error::Io(e.to_string()))?;
280        let (verified, facets, max_items, ready) = match checked {
281            Ok(value) => value,
282            Err(error) => {
283                writer
284                    .fail(&api::heddle::api::common::CallFailure {
285                        code: 7,
286                        message: error.to_string(),
287                        ..Default::default()
288                    })
289                    .await?;
290                return Err(error.into());
291            }
292        };
293        let negotiated_protocol = ready.protocol.clone();
294        writer
295            .send(
296                ReplicateThreadResponse {
297                    body: Some(replicate_thread_response::Body::Ready(ready)),
298                }
299                .encode_to_vec(),
300            )
301            .await?;
302        let session = Session::new(self.local_replica(store), remote_key, facets, max_items)?
303            .with_protocol(negotiated_protocol.as_ref(), negotiated_protocol.as_ref())?;
304        live_replication::run(
305            session,
306            reader,
307            writer,
308            Side::Acceptor,
309            feed,
310            move |activity| {
311                std::future::ready(
312                    if matches!(activity, live_replication::Activity::InputConsumed { .. }) {
313                        Ok(())
314                    } else {
315                        authority.recheck(&verified)
316                    },
317                )
318            },
319        )
320        .await
321    }
322}
323
324fn io_error(error: impl std::fmt::Display) -> transport::Error {
325    transport::Error::Io(error.to_string())
326}
327fn store_error(error: repo::thread_replication::Error) -> transport::Error {
328    io_error(error)
329}
330
331#[cfg(test)]
332#[path = "replication_rpc_tests.rs"]
333mod tests;
heddle_thread_api/replication_rpc.rs

heddle_thread_api/
replication_rpc.rs
