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            thread: Some(thread.clone()),
120            facets: self
121                .facets
122                .iter()
123                .copied()
124                .map(replication::wire_facet)
125                .collect(),
126            sharing_policy_version: version.map(|v| v.as_bytes().to_vec()).unwrap_or_default(),
127            thread_genesis: Some(self.genesis.clone()),
128            budget: Some(ReadBudget {
129                max_items: 64,
130                max_frame_bytes: FRAME_LIMIT as u32,
131                max_snapshot_bytes: 0,
132            }),
133            record_formats: vec![OPERATION_FORMAT.into()],
134            session_nonce: uuid::Uuid::new_v4().as_bytes().to_vec(),
135            source: Some(self.endpoint.clone()),
136            destination: Some(destination.clone()),
137            // No HYBRID import-authority support is claimed; no proof bundle.
138            protocol: None,
139            import_authority: None,
140        };
141        let transport = IrohTransport::new(connection, signer, FRAME_LIMIT, TIMEOUT)?;
142        let (writer, mut reader) = transport
143            .exchange(
144                rpc::SyncServiceReplicateThread::METHOD,
145                ReplicateThreadRequest {
146                    body: Some(replicate_thread_request::Body::Open(opening)),
147                }
148                .encode_to_vec(),
149            )
150            .await?;
151        let bytes = reader.next().await?.ok_or(transport::Error::Protocol(
152            "replication closed before Ready",
153        ))?;
154        let response =
155            ReplicateThreadResponse::decode(bytes.as_slice()).map_err(transport::Error::from)?;
156        let Some(replicate_thread_response::Body::Ready(ready)) = response.body else {
157            return Err(transport::Error::Protocol("replication requires Ready").into());
158        };
159        let (facets, max_items) =
160            opening::validate_ready(&ready, &thread, &destination, &self.facets, 64)?;
161        let session = Session::new(self.local_replica(store), remote_key, facets, max_items)?;
162        live_replication::run(
163            session,
164            reader,
165            writer,
166            Side::Initiator,
167            feed,
168            move |activity| {
169                let authorize = authorize.clone();
170                async move {
171                    if matches!(activity, live_replication::Activity::InputConsumed { .. }) {
172                        Ok(())
173                    } else {
174                        authorize().await
175                    }
176                }
177            },
178        )
179        .await
180    }
181
182    /// Accept exactly one ReplicateThread RPC. The connection must be routed
183    /// here by the host; other service methods are not advertised or emulated.
184    pub async fn accept<S>(
185        &self,
186        connection: Connection,
187        authority: Arc<RootAuthority>,
188        store: Arc<S>,
189        feed: &Feed,
190    ) -> live_replication::Result<(), replication::native::Error>
191    where
192        S: ObjectStore + Send + Sync + 'static,
193    {
194        let remote_key = *connection.remote_id().as_bytes();
195        let (send, mut recv) = tokio::time::timeout(TIMEOUT, connection.accept_bi())
196            .await
197            .map_err(|_| transport::Error::Timeout)?
198            .map_err(io_error)?;
199        let mut writer = Writer::new(send, FRAME_LIMIT, TIMEOUT);
200        let opening = async {
201            let mut prelude = vec![0; 6];
202            recv.read_exact(&mut prelude).await.map_err(io_error)?;
203            let method_size = u16::from_be_bytes([prelude[0], prelude[1]]) as usize;
204            let context_size =
205                u32::from_be_bytes([prelude[2], prelude[3], prelude[4], prelude[5]]) as usize;
206            if method_size == 0
207                || method_size > framing::MAX_METHOD_PATH
208                || context_size > framing::MAX_CALL_CONTEXT
209            {
210                return Err(transport::Error::Protocol(
211                    "opening exceeds metadata limits",
212                ));
213            }
214            prelude.resize(6 + method_size + context_size, 0);
215            recv.read_exact(&mut prelude[6..]).await.map_err(io_error)?;
216            let (request, _) = framing::decode_request_prelude(&prelude)?
217                .ok_or(transport::Error::Protocol("incomplete opening metadata"))?;
218            if request.method != rpc::SyncServiceReplicateThread::METHOD.path {
219                return Err(transport::Error::Protocol(
220                    "unsupported RPC on replication boundary",
221                ));
222            }
223            Ok(request.context)
224        };
225        let context = tokio::time::timeout(TIMEOUT, opening)
226            .await
227            .map_err(|_| transport::Error::Timeout)??;
228        let mut reader = Reader::for_method(
229            recv,
230            FRAME_LIMIT,
231            TIMEOUT,
232            rpc::SyncServiceReplicateThread::METHOD,
233        );
234        let bytes = reader
235            .next()
236            .await?
237            .ok_or(transport::Error::Protocol("missing replication opening"))?;
238        let peer = self.clone();
239        let verifier = authority.clone();
240        let checked = tokio::task::spawn_blocking(move || -> Result<_, transport::Error> {
241            let request = ReplicateThreadRequest::decode(bytes.as_slice())?;
242            let Some(replicate_thread_request::Body::Open(open)) = request.body else {
243                return Err(transport::Error::Protocol("replication requires Open"));
244            };
245            let (_, version) = peer.replica.sharing(&remote_key).map_err(store_error)?;
246            let ready = opening::accept(
247                &open,
248                &peer.reference()?,
249                &peer.endpoint,
250                remote_key,
251                &peer.facets,
252                version.map(|v| v.as_bytes().to_vec()).unwrap_or_default(),
253            )?
254            .ready;
255            let (facets, max_items) = opening::validate_ready(
256                &ready,
257                &peer.reference()?,
258                &peer.endpoint,
259                &peer.facets,
260                64,
261            )?;
262            let verified = verifier.verify(
263                &context,
264                rpc::SyncServiceReplicateThread::METHOD,
265                &bytes,
266                "write",
267                &peer.replica,
268            )?;
269            Ok((verified, facets, max_items, ready))
270        })
271        .await
272        .map_err(|e| transport::Error::Io(e.to_string()))?;
273        let (verified, facets, max_items, ready) = match checked {
274            Ok(value) => value,
275            Err(error) => {
276                writer
277                    .fail(&api::heddle::api::common::CallFailure {
278                        code: 7,
279                        message: error.to_string(),
280                        ..Default::default()
281                    })
282                    .await?;
283                return Err(error.into());
284            }
285        };
286        writer
287            .send(
288                ReplicateThreadResponse {
289                    body: Some(replicate_thread_response::Body::Ready(ready)),
290                }
291                .encode_to_vec(),
292            )
293            .await?;
294        let session = Session::new(self.local_replica(store), remote_key, facets, max_items)?;
295        live_replication::run(
296            session,
297            reader,
298            writer,
299            Side::Acceptor,
300            feed,
301            move |activity| {
302                std::future::ready(
303                    if matches!(activity, live_replication::Activity::InputConsumed { .. }) {
304                        Ok(())
305                    } else {
306                        authority.recheck(&verified)
307                    },
308                )
309            },
310        )
311        .await
312    }
313}
314
315fn io_error(error: impl std::fmt::Display) -> transport::Error {
316    transport::Error::Io(error.to_string())
317}
318fn store_error(error: repo::thread_replication::Error) -> transport::Error {
319    io_error(error)
320}
321
322#[cfg(test)]
323#[path = "replication_rpc_tests.rs"]
324mod tests;
heddle_thread_api/replication_rpc.rs

heddle_thread_api/
replication_rpc.rs
