heddle_thread_api/
replication.rs

1// SPDX-License-Identifier: Apache-2.0
2//! A bounded, transport-neutral causal exchange shared by device and hosted
3//! endpoints. The RPC adapter supplies the verified request scope.
4use std::collections::BTreeSet;
5
6use crypto::thread_operation::SignedOperation;
7use heddle_object_model::object::{
8    ContentHash,
9    thread_replication::{Admission, OPERATION_FORMAT, ThreadFacet},
10};
11#[cfg(feature = "native")]
12pub mod native;
13pub mod opening;
14pub mod ownership;
15pub mod store;
16use store::ReplicaStore;
17
18use crate::contract::*;
19
20#[derive(Debug, thiserror::Error)]
21pub enum Error {
22    #[error(transparent)]
23    Operation(#[from] heddle_object_model::error::HeddleError),
24    #[error(transparent)]
25    Signature(#[from] crypto::thread_operation::Error),
26    #[error("replication protocol: {0}")]
27    Protocol(&'static str),
28}
29pub type Result<T> = std::result::Result<T, Error>;
30
31#[derive(Debug, thiserror::Error)]
32pub enum StoreError<E: std::error::Error + 'static> {
33    #[error("replica store: {0}")]
34    Store(#[source] E),
35    #[error(transparent)]
36    Protocol(#[from] Error),
37}
38pub type StoreResult<T, E> = std::result::Result<T, StoreError<E>>;
39
40/// Internal frames carry the same typed payloads in both directions.
41pub enum Frame {
42    Have(ReplicationHave),
43    Need(ReplicationNeed),
44    Operations(ReplicationOperations),
45    Receipt(ReplicationReceipt),
46}
47/// Validated originals retain shared evidence without re-encoding a wire batch
48/// for every committed-prefix unit.
49pub(crate) enum InputUnit {
50    Frame(Frame),
51    Operation(store::ReceivedOperation),
52}
53impl Frame {
54    pub fn request(self) -> ReplicateThreadRequest {
55        use replicate_thread_request::Body;
56        ReplicateThreadRequest {
57            body: Some(match self {
58                Self::Have(v) => Body::Have(v),
59                Self::Need(v) => Body::Need(v),
60                Self::Operations(v) => Body::Operations(v),
61                Self::Receipt(v) => Body::Receipt(v),
62            }),
63        }
64    }
65    pub fn response(self) -> ReplicateThreadResponse {
66        use replicate_thread_response::Body;
67        ReplicateThreadResponse {
68            body: Some(match self {
69                Self::Have(v) => Body::Have(v),
70                Self::Need(v) => Body::Need(v),
71                Self::Operations(v) => Body::Operations(v),
72                Self::Receipt(v) => Body::Receipt(v),
73            }),
74        }
75    }
76    pub fn from_request(request: ReplicateThreadRequest) -> Result<Self> {
77        use replicate_thread_request::Body;
78        Ok(match request.body {
79            Some(Body::Have(v)) => Self::Have(v),
80            Some(Body::Need(v)) => Self::Need(v),
81            Some(Body::Operations(v)) => Self::Operations(v),
82            Some(Body::Receipt(v)) => Self::Receipt(v),
83            _ => return Err(Error::Protocol("unexpected replication opening")),
84        })
85    }
86    pub fn from_response(response: ReplicateThreadResponse) -> Result<Self> {
87        use replicate_thread_response::Body;
88        Ok(match response.body {
89            Some(Body::Have(v)) => Self::Have(v),
90            Some(Body::Need(v)) => Self::Need(v),
91            Some(Body::Operations(v)) => Self::Operations(v),
92            Some(Body::Receipt(v)) => Self::Receipt(v),
93            _ => return Err(Error::Protocol("unexpected replication ready")),
94        })
95    }
96}
97
98pub enum Outbound {
99    Frame(Frame),
100    Operation(ContentHash),
101}
102
103#[derive(Clone)]
104pub struct Session<B: ReplicaStore> {
105    pub replica: B,
106    destination: [u8; 32],
107    facets: BTreeSet<ThreadFacet>,
108    max_items: usize,
109    in_flight: BTreeSet<ContentHash>,
110    generation: i64,
111    announce_generation: i64,
112    announce_facet: usize,
113    after: Option<ContentHash>,
114    pending_input_bookkeeping: Option<(ThreadFacet, ContentHash)>,
115}
116impl<B: ReplicaStore> Session<B> {
117    /// `facets` is the intersection of authenticated admission scope and the
118    /// negotiated opening. Local export still checks current sharing policy.
119    pub fn new(
120        replica: B,
121        destination: [u8; 32],
122        facets: BTreeSet<ThreadFacet>,
123        max_items: usize,
124    ) -> Result<Self> {
125        if max_items == 0 || max_items > 64 {
126            return Err(Error::Protocol("replication batch must be 1..64"));
127        }
128        Ok(Self {
129            replica,
130            destination,
131            facets,
132            max_items,
133            in_flight: BTreeSet::new(),
134            generation: -1,
135            announce_generation: -1,
136            announce_facet: 0,
137            after: None,
138            pending_input_bookkeeping: None,
139        })
140    }
141    pub async fn export_facets(&self) -> StoreResult<BTreeSet<ThreadFacet>, B::Error> {
142        let sharing = self
143            .replica
144            .sharing(self.destination)
145            .await
146            .map_err(StoreError::Store)?;
147        Ok(sharing.intersection(&self.facets).copied().collect())
148    }
149    /// At most one bounded frontier page. Call again until None. A generation
150    /// change during a paged announcement starts another round, closing gaps.
151    pub async fn announcement(&mut self) -> StoreResult<Option<Frame>, B::Error> {
152        let current = self.replica.generation().await.map_err(StoreError::Store)?;
153        if self.generation == current && self.announce_generation < 0 {
154            return Ok(None);
155        }
156        if self.announce_generation < 0 {
157            self.announce_generation = current;
158            self.announce_facet = 0;
159            self.after = None;
160        }
161        let sharing = self.export_facets().await?;
162        let facets = ThreadFacet::ALL;
163        while self.announce_facet < facets.len() {
164            let facet = facets[self.announce_facet];
165            if !sharing.contains(&facet) {
166                self.announce_facet += 1;
167                self.after = None;
168                continue;
169            }
170            let page = self
171                .replica
172                .frontier_page(facet, self.after, self.max_items)
173                .await
174                .map_err(StoreError::Store)?;
175            if page.is_empty() {
176                self.announce_facet += 1;
177                self.after = None;
178                continue;
179            }
180            self.after = page.last().copied();
181            return Ok(Some(Frame::Have(ReplicationHave {
182                frontiers: vec![CausalFrontier {
183                    facet: wire_facet(facet),
184                    heads: page.into_iter().map(|id| id.as_bytes().to_vec()).collect(),
185                }],
186            })));
187        }
188        self.generation = self.announce_generation;
189        self.announce_generation = -1;
190        // A watcher may already have delivered a mutation that happened while
191        // this round was being paged. Keep the caller draining until a fresh
192        // round covers it; None must mean the announcement is caught up.
193        Ok(
194            (self.replica.generation().await.map_err(StoreError::Store)? != self.generation)
195                .then(|| Frame::Have(ReplicationHave::default())),
196        )
197    }
198    pub async fn handle(&mut self, frame: Frame) -> StoreResult<Vec<Outbound>, B::Error> {
199        self.handle_frame(frame, true).await
200    }
201    /// Validate the complete wire batch before committing any prefix. Each
202    /// returned unit can then receive its own durable acknowledgement.
203    pub(crate) fn input_units(&self, frame: Frame) -> StoreResult<Vec<InputUnit>, B::Error> {
204        let Frame::Operations(batch) = frame else {
205            return Ok(vec![InputUnit::Frame(frame)]);
206        };
207        self.check_count(batch.operations.len())?;
208        self.check_count(batch.authority_admissions.len())?;
209        self.check_count(batch.boundary_acceptances.len())?;
210        let originals = crate::authority_admission::match_batch(&batch)
211            .map_err(|_| Error::Protocol("invalid original authority batch"))?;
212        let mut units = Vec::with_capacity(originals.len());
213        for received in originals {
214            let operation = received.original.verify().map_err(Error::from)?;
215            if !self.facets.contains(&operation.facet()) {
216                return Err(Error::Protocol("operation is outside admission scope").into());
217            }
218            units.push(InputUnit::Operation(received));
219        }
220        Ok(units)
221    }
222    pub(crate) async fn handle_unit(
223        &mut self,
224        unit: InputUnit,
225    ) -> StoreResult<Vec<Outbound>, B::Error> {
226        match unit {
227            InputUnit::Frame(frame) => self.handle_input(frame).await,
228            InputUnit::Operation(received) => Ok(vec![Outbound::Frame(Frame::Receipt(
229                self.commit_original(received, false).await?,
230            ))]),
231        }
232    }
233    async fn commit_original(
234        &mut self,
235        received: store::ReceivedOperation,
236        maintenance: bool,
237    ) -> StoreResult<ReplicationReceipt, B::Error> {
238        if self.pending_input_bookkeeping.is_some() {
239            return Err(Error::Protocol("prior input bookkeeping not completed").into());
240        }
241        let operation = received.original.verify().map_err(Error::from)?;
242        if !self.facets.contains(&operation.facet()) {
243            return Err(Error::Protocol("operation is outside admission scope").into());
244        }
245        let id = operation.id().map_err(Error::from)?;
246        self.in_flight.remove(&id);
247        let mut receipt = ReplicationReceipt::default();
248        match self
249            .replica
250            .receive(received)
251            .await
252            .map_err(StoreError::Store)?
253        {
254            Admission::Accepted => receipt.accepted_operation_ids.push(id.as_bytes().to_vec()),
255            Admission::Pending => {
256                receipt.pending_operation_ids.push(id.as_bytes().to_vec());
257                if maintenance {
258                    self.replica
259                        .remember_peer_heads(self.destination, vec![(operation.facet(), id)])
260                        .await
261                        .map_err(StoreError::Store)?;
262                } else {
263                    self.pending_input_bookkeeping = Some((operation.facet(), id));
264                }
265            }
266            Admission::Rejected(message) => receipt.rejected.push(rejection(id, message)),
267        }
268        Ok(receipt)
269    }
270    pub(crate) fn has_input_bookkeeping(&self) -> bool {
271        self.pending_input_bookkeeping.is_some()
272    }
273    pub(crate) async fn finish_input_bookkeeping(&mut self) -> StoreResult<(), B::Error> {
274        if let Some(head) = self.pending_input_bookkeeping.take() {
275            self.replica
276                .remember_peer_heads(self.destination, vec![head])
277                .await
278                .map_err(StoreError::Store)?;
279        }
280        Ok(())
281    }
282    /// Return the committed result before optional peer repair bookkeeping.
283    pub(crate) async fn handle_input(
284        &mut self,
285        frame: Frame,
286    ) -> StoreResult<Vec<Outbound>, B::Error> {
287        self.handle_frame(frame, false).await
288    }
289    async fn handle_frame(
290        &mut self,
291        frame: Frame,
292        maintenance: bool,
293    ) -> StoreResult<Vec<Outbound>, B::Error> {
294        let mut responses = Vec::new();
295        match frame {
296            Frame::Have(have) => {
297                let count: usize = have.frontiers.iter().map(|f| f.heads.len()).sum();
298                self.check_count(count)?;
299                self.check_count(have.frontiers.len())?;
300                let mut heads = Vec::new();
301                let mut receipt = ReplicationReceipt::default();
302                for frontier in have.frontiers {
303                    let facet = native_facet(frontier.facet)?;
304                    if !self.facets.contains(&facet) {
305                        return Err(Error::Protocol("unnegotiated facet").into());
306                    }
307                    for bytes in frontier.heads {
308                        let id = hash(&bytes)?;
309                        if let Some((record, status)) = self
310                            .replica
311                            .operation(id)
312                            .await
313                            .map_err(StoreError::Store)?
314                        {
315                            if record.original.verify().map_err(Error::from)?.facet() != facet {
316                                return Err(Error::Protocol(
317                                    "advertised frontier has the wrong facet",
318                                )
319                                .into());
320                            }
321                            match status {
322                                Admission::Accepted => receipt.accepted_operation_ids.push(bytes),
323                                Admission::Rejected(message) => {
324                                    receipt.rejected.push(rejection(id, message))
325                                }
326                                Admission::Pending => heads.push((facet, id)),
327                            }
328                        } else {
329                            heads.push((facet, id));
330                        }
331                    }
332                }
333                self.replica
334                    .remember_peer_heads(self.destination, heads)
335                    .await
336                    .map_err(StoreError::Store)?;
337                if !receipt.accepted_operation_ids.is_empty() || !receipt.rejected.is_empty() {
338                    responses.push(Outbound::Frame(Frame::Receipt(receipt)));
339                }
340            }
341            Frame::Need(need) => {
342                self.check_count(need.operation_ids.len())?;
343                let sharing = self.export_facets().await?;
344                for bytes in need.operation_ids {
345                    let id = hash(&bytes)?;
346                    let Some((record, Admission::Accepted)) = self
347                        .replica
348                        .operation(id)
349                        .await
350                        .map_err(StoreError::Store)?
351                    else {
352                        return Err(Error::Protocol("requested operation unavailable").into());
353                    };
354                    let operation = record.original.verify().map_err(Error::from)?;
355                    if !sharing.contains(&operation.facet()) {
356                        return Err(
357                            Error::Protocol("operation is outside current sharing policy").into(),
358                        );
359                    }
360                    responses.push(Outbound::Operation(id));
361                }
362            }
363            Frame::Operations(batch) => {
364                if self.pending_input_bookkeeping.is_some() {
365                    return Err(Error::Protocol("prior input bookkeeping not completed").into());
366                }
367                self.check_count(batch.operations.len())?;
368                self.check_count(batch.authority_admissions.len())?;
369                self.check_count(batch.boundary_acceptances.len())?;
370                let decoded =
371                    crate::authority_admission::match_batch(&batch).map_err(
372                        |error| match error {
373                            crate::transport::Error::Protocol(message) => Error::Protocol(message),
374                            _ => Error::Protocol("invalid original authority batch"),
375                        },
376                    )?;
377                let mut operations = Vec::with_capacity(decoded.len());
378                for received in decoded {
379                    let operation = received.original.verify().map_err(Error::from)?;
380                    let id = operation.id().map_err(Error::from)?;
381                    if !self.facets.contains(&operation.facet()) {
382                        return Err(Error::Protocol("operation is outside admission scope").into());
383                    }
384                    operations.push((received, operation, id));
385                }
386                let mut receipt = ReplicationReceipt::default();
387                for (received, _, _) in operations {
388                    let next = self.commit_original(received, maintenance).await?;
389                    receipt
390                        .accepted_operation_ids
391                        .extend(next.accepted_operation_ids);
392                    receipt
393                        .pending_operation_ids
394                        .extend(next.pending_operation_ids);
395                    receipt.rejected.extend(next.rejected);
396                }
397                responses.push(Outbound::Frame(Frame::Receipt(receipt)));
398            }
399            Frame::Receipt(receipt) => {
400                self.check_count(
401                    receipt.accepted_operation_ids.len()
402                        + receipt.pending_operation_ids.len()
403                        + receipt.rejected.len(),
404                )?;
405                for bytes in receipt.accepted_operation_ids {
406                    self.replica
407                        .record_peer_receipt(self.destination, hash(&bytes)?, Admission::Accepted)
408                        .await
409                        .map_err(StoreError::Store)?;
410                }
411                for bytes in receipt.pending_operation_ids {
412                    self.replica
413                        .record_peer_receipt(self.destination, hash(&bytes)?, Admission::Pending)
414                        .await
415                        .map_err(StoreError::Store)?;
416                }
417                for rejected in receipt.rejected {
418                    self.replica
419                        .record_peer_receipt(
420                            self.destination,
421                            hash(&rejected.operation_id)?,
422                            Admission::Rejected(
423                                rejected.failure.map(|f| f.message).unwrap_or_default(),
424                            ),
425                        )
426                        .await
427                        .map_err(StoreError::Store)?;
428                }
429            }
430        }
431        if maintenance && let Some(repair) = self.control().await? {
432            responses.push(Outbound::Frame(repair));
433        }
434        Ok(responses)
435    }
436
437    /// One bounded dependency window, refilled after each received operation.
438    /// Peer frontiers are durable; outstanding requests are session-local.
439    pub async fn control(&mut self) -> StoreResult<Option<Frame>, B::Error> {
440        let settled = self
441            .replica
442            .settled_peer_heads(self.destination, self.facets.clone(), self.max_items)
443            .await
444            .map_err(StoreError::Store)?;
445        if !settled.is_empty() {
446            let mut receipt = ReplicationReceipt::default();
447            for (id, admission) in settled {
448                self.in_flight.remove(&id);
449                match admission {
450                    Admission::Accepted => {
451                        receipt.accepted_operation_ids.push(id.as_bytes().to_vec())
452                    }
453                    Admission::Rejected(message) => receipt.rejected.push(rejection(id, message)),
454                    Admission::Pending => {}
455                }
456            }
457            return Ok(Some(Frame::Receipt(receipt)));
458        }
459        let available = self.max_items.saturating_sub(self.in_flight.len());
460        if available == 0 {
461            return Ok(None);
462        }
463        let candidates = self
464            .replica
465            .needed_from_peer(self.destination, self.facets.clone(), self.max_items)
466            .await
467            .map_err(StoreError::Store)?;
468        let ids: Vec<_> = candidates
469            .into_iter()
470            .filter(|id| !self.in_flight.contains(id))
471            .take(available)
472            .collect();
473        self.in_flight.extend(ids.iter().copied());
474        Ok((!ids.is_empty()).then(|| {
475            Frame::Need(ReplicationNeed {
476                operation_ids: ids.into_iter().map(|id| id.as_bytes().to_vec()).collect(),
477            })
478        }))
479    }
480
481    /// Resolve payload only when the writer is ready. Pending output queues
482    /// contain IDs, and policy is checked again immediately before disclosure.
483    pub async fn export_operation(&self, id: ContentHash) -> StoreResult<Frame, B::Error> {
484        let Some((record, Admission::Accepted)) = self
485            .replica
486            .operation(id)
487            .await
488            .map_err(StoreError::Store)?
489        else {
490            return Err(Error::Protocol("requested operation unavailable").into());
491        };
492        let operation = record.original.verify().map_err(Error::from)?;
493        if !self.export_facets().await?.contains(&operation.facet()) {
494            return Err(Error::Protocol("operation is outside current sharing policy").into());
495        }
496        Ok(Frame::Operations(ReplicationOperations {
497            boundary_acceptances: crate::boundary_acceptance::authority_evidence(
498                record.authority_admission.as_ref(),
499            )
500            .map_err(|_| Error::Protocol("invalid boundary evidence"))?,
501            authority_admissions: record
502                .authority_admission
503                .as_ref()
504                .map(crate::authority_admission::encode)
505                .transpose()
506                .map_err(|_| Error::Protocol("invalid retained authority admission"))?
507                .into_iter()
508                .collect(),
509            operations: vec![SignedRecord {
510                format: OPERATION_FORMAT.into(),
511                canonical_record: record.original.canonical,
512                signatures: vec![RecordSignature {
513                    public_key: operation.publisher.to_vec(),
514                    signature: record.original.signature,
515                }],
516            }],
517        }))
518    }
519
520    fn check_count(&self, count: usize) -> Result<()> {
521        if count > self.max_items {
522            Err(Error::Protocol("replication item budget exceeded"))
523        } else {
524            Ok(())
525        }
526    }
527}
528
529/// Exact completion fields for one durably admitted original operation.
530/// Hosts can bind their acknowledgement exception to these canonical fields.
531pub fn completion_receipt(id: ContentHash, admission: Admission) -> ReplicationReceipt {
532    let mut receipt = ReplicationReceipt::default();
533    match admission {
534        Admission::Accepted => receipt.accepted_operation_ids.push(id.as_bytes().to_vec()),
535        Admission::Pending => receipt.pending_operation_ids.push(id.as_bytes().to_vec()),
536        Admission::Rejected(message) => receipt.rejected.push(rejection(id, message)),
537    }
538    receipt
539}
540
541fn rejection(id: ContentHash, message: String) -> ReplicationRejection {
542    ReplicationRejection {
543        operation_id: id.as_bytes().to_vec(),
544        failure: Some(api::heddle::api::common::CallFailure {
545            code: 9,
546            message,
547            ..Default::default()
548        }),
549    }
550}
551
552pub fn decode_record(record: SignedRecord) -> Result<SignedOperation> {
553    if record.format != OPERATION_FORMAT || record.signatures.len() != 1 {
554        return Err(Error::Protocol("unsupported signed operation format"));
555    }
556    let signature = &record.signatures[0];
557    let signed = SignedOperation {
558        canonical: record.canonical_record,
559        signature: signature.signature.clone(),
560    };
561    if signed.verify()?.publisher.as_slice() != signature.public_key {
562        return Err(Error::Protocol(
563            "record signature key differs from publisher",
564        ));
565    }
566    Ok(signed)
567}
568pub fn native_facet(facet: i32) -> Result<ThreadFacet> {
569    match SharedFacet::try_from(facet) {
570        Ok(SharedFacet::Source) => Ok(ThreadFacet::Source),
571        Ok(SharedFacet::Collaboration) => Ok(ThreadFacet::Discussion),
572        Ok(SharedFacet::Metadata) => Ok(ThreadFacet::Metadata),
573        _ => Err(Error::Protocol("unsupported replication facet")),
574    }
575}
576pub fn wire_facet(facet: ThreadFacet) -> i32 {
577    match facet {
578        ThreadFacet::Source => SharedFacet::Source as i32,
579        ThreadFacet::Discussion => SharedFacet::Collaboration as i32,
580        ThreadFacet::Metadata => SharedFacet::Metadata as i32,
581    }
582}
583fn hash(bytes: &[u8]) -> Result<ContentHash> {
584    Ok(ContentHash::from_bytes(bytes.try_into().map_err(|_| {
585        Error::Protocol("operation ID must be 32 bytes")
586    })?))
587}
588
589#[cfg(all(test, feature = "native"))]
590#[path = "replication_tests.rs"]
591mod tests;
592
593#[cfg(all(test, feature = "native"))]
594#[path = "replication_admission_tests.rs"]
595mod admission_tests;
heddle_thread_api/replication.rs

heddle_thread_api/
replication.rs
