heddle_thread_api/replication/
native.rs

1//! Local SQLite adapter. Blocking work stays off the stream runtime.
2use std::{collections::BTreeSet, path::PathBuf, sync::Arc};
3
4use heddle_object_model::object::{
5    ContentHash,
6    thread_replication::{Admission, ThreadFacet},
7};
8use objects::store::ObjectStore;
9use repo::thread_replication::ThreadReplica;
10
11use super::store::{ReceivedOperation, ReplicaStore};
12
13#[derive(Debug, thiserror::Error)]
14pub enum Error {
15    #[error(transparent)]
16    Store(#[from] repo::thread_replication::Error),
17    #[error("local replica worker: {0}")]
18    Worker(#[from] tokio::task::JoinError),
19}
20pub struct LocalReplica<S> {
21    replica: ThreadReplica,
22    objects: Arc<S>,
23    authority_home: Option<PathBuf>,
24}
25impl<S> Clone for LocalReplica<S> {
26    fn clone(&self) -> Self {
27        Self {
28            replica: self.replica.clone(),
29            objects: self.objects.clone(),
30            authority_home: self.authority_home.clone(),
31        }
32    }
33}
34impl<S: ObjectStore + Send + Sync + 'static> LocalReplica<S> {
35    pub fn new(replica: ThreadReplica, objects: Arc<S>) -> Self {
36        Self {
37            replica,
38            objects,
39            authority_home: None,
40        }
41    }
42    /// Use independently enrolled local account authority for original metadata
43    /// authors. Without this binding, new metadata admission fails closed.
44    pub fn with_device_authority(mut self, home: PathBuf) -> Self {
45        self.authority_home = Some(home);
46        self
47    }
48    async fn execute<T: Send + 'static>(
49        &self,
50        operation: impl FnOnce(&ThreadReplica, &S) -> repo::thread_replication::Result<T>
51        + Send
52        + 'static,
53    ) -> Result<T, Error> {
54        let local = self.clone();
55        Ok(
56            tokio::task::spawn_blocking(move || operation(&local.replica, &local.objects))
57                .await??,
58        )
59    }
60}
61impl<S: ObjectStore + Send + Sync + 'static> ReplicaStore for LocalReplica<S> {
62    type Error = Error;
63    fn thread_id(&self) -> ContentHash {
64        self.replica.thread_id()
65    }
66    async fn generation(&self) -> Result<i64, Error> {
67        self.execute(|replica, _| replica.generation()).await
68    }
69    async fn sharing(&self, destination: [u8; 32]) -> Result<BTreeSet<ThreadFacet>, Error> {
70        self.execute(move |replica, _| replica.sharing(&destination).map(|(facets, _)| facets))
71            .await
72    }
73    async fn frontier_page(
74        &self,
75        facet: ThreadFacet,
76        after: Option<ContentHash>,
77        limit: usize,
78    ) -> Result<Vec<ContentHash>, Error> {
79        self.execute(move |replica, _| replica.frontier_page(facet, after, limit))
80            .await
81    }
82    async fn operation(
83        &self,
84        id: ContentHash,
85    ) -> Result<Option<(ReceivedOperation, Admission)>, Error> {
86        self.execute(move |replica, _| {
87            Ok(replica
88                .operation_with_authority_admission(&id)?
89                .map(|stored| {
90                    (
91                        ReceivedOperation {
92                            original: stored.original,
93                            authority_admission: stored.authority_admission,
94                        },
95                        stored.status,
96                    )
97                }))
98        })
99        .await
100    }
101    async fn receive(&self, received: ReceivedOperation) -> Result<Admission, Error> {
102        let authority_home = self.authority_home.clone();
103        self.execute(move |replica, objects| {
104            let operation = received.original;
105            if let Some(receipt) = received.authority_admission {
106                return replica.receive_with_authority_admission(&operation, &receipt, objects, |_| Ok(()));
107            }
108            replica.receive(&operation, objects, |native| {
109                use heddle_object_model::object::thread_replication::ThreadOperationBody;
110                if !matches!(native.body, ThreadOperationBody::Metadata(_)) && native.source_author()?.is_none() {
111                    return Ok(());
112                }
113                // Durable original-author receipt remains valid while causal
114                // parents arrive later. Neither the envelope nor claimed time
115                // can synthesize the atomically retained admission marker.
116                if replica.original_authority_admitted(&operation)? {
117                    return Ok(());
118                }
119                let genesis = replica.genesis()?;
120                if matches!(native.source_author()?, Some(heddle_object_model::object::thread_replication::SourceAuthor::LocalKey)) {
121                    return replica.verify_local_source_owner(native);
122                }
123                let home = authority_home.as_ref().ok_or_else(|| {
124                    repo::thread_replication::Error::Invalid(
125                        "original operation requires independently enrolled account authority"
126                            .into(),
127                    )
128                })?;
129                let now = chrono::Utc::now().timestamp();
130                let authority = repo::device_authority::load(home, now).map_err(authority_error)?;
131                if let ThreadOperationBody::Metadata(bytes) = &native.body {
132                    heddle_object_model::object::thread_replication::metadata::ThreadControl::decode(bytes)?.validate_parents(&genesis, &[])?;
133                }
134                let spool = genesis.spool.parse().map_err(authority_error)?;
135                let registered =
136                    repo::device_catalog::load(home, spool).map_err(authority_error)?;
137                if native.source_author()?.is_some() {
138                    replica.verify_source_authority(native, &authority, &registered.capability_path, now)
139                } else {
140                    repo::thread_replication::metadata::verify_control_authority(native, &authority, &registered.capability_path, now)
141                }
142
143            })
144        })
145        .await
146    }
147    async fn remember_peer_heads(
148        &self,
149        peer: [u8; 32],
150        heads: Vec<(ThreadFacet, ContentHash)>,
151    ) -> Result<(), Error> {
152        self.execute(move |replica, _| replica.remember_peer_heads(peer, &heads))
153            .await
154    }
155    async fn record_peer_receipt(
156        &self,
157        peer: [u8; 32],
158        id: ContentHash,
159        admission: Admission,
160    ) -> Result<(), Error> {
161        self.execute(move |replica, _| replica.record_peer_receipt(peer, id, &admission))
162            .await
163    }
164    async fn settled_peer_heads(
165        &self,
166        peer: [u8; 32],
167        facets: BTreeSet<ThreadFacet>,
168        limit: usize,
169    ) -> Result<Vec<(ContentHash, Admission)>, Error> {
170        self.execute(move |replica, _| replica.settled_peer_heads(peer, &facets, limit))
171            .await
172    }
173    async fn needed_from_peer(
174        &self,
175        peer: [u8; 32],
176        facets: BTreeSet<ThreadFacet>,
177        limit: usize,
178    ) -> Result<Vec<ContentHash>, Error> {
179        self.execute(move |replica, _| replica.needed_from_peer(peer, &facets, limit))
180            .await
181    }
182}
183
184fn authority_error(error: impl std::fmt::Display) -> repo::thread_replication::Error {
185    repo::thread_replication::Error::Invalid(error.to_string())
186}
heddle_thread_api/replication/native.rs

heddle_thread_api/replication/
native.rs
