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

heddle_thread_api/
replication_rpc.rs
