add resolution strategy

Author: insipx

xmtp_api_d14n/src/protocol/mod.rs

@@ -11,7 +11,9 @@ pub use in_memory_cursor_store::*;
 
 mod impls;
 
-mod resolution;
-// pub use resolution::*;
+pub mod resolve;
 
 pub mod sort;
+
+mod order;
+pub use order::*;

xmtp_api_d14n/src/protocol/order.rs

@@ -0,0 +1,47 @@
+use crate::protocol::{
+    Envelope, EnvelopeError, OrderedEnvelopeCollection, ResolveDependencies, Sort, sort,
+};
+use derive_builder::Builder;
+use xmtp_proto::types::TopicCursor;
+
+#[derive(Debug, Clone, Builder)]
+#[builder(setter(strip_option), build_fn(error = "EnvelopeError"))]
+pub struct Ordered<T, R> {
+    envelopes: Vec<T>,
+    resolver: R,
+}
+
+impl<T: Clone, R: Clone> Ordered<T, R> {
+    pub fn builder() -> OrderedBuilder<T, R> {
+        OrderedBuilder::default()
+    }
+}
+
+#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
+impl<T, R> OrderedEnvelopeCollection for Ordered<T, R>
+where
+    T: Envelope<'static>,
+    R: ResolveDependencies<ResolvedEnvelope = T>,
+{
+    /// Sort dependencies of `Self` according to XIP
+    /// TODO fill in docs
+    async fn sort(&mut self) -> Result<(), EnvelopeError> {
+        let mut topic_cursor = TopicCursor::default();
+        {
+            sort::timestamp(&mut self.envelopes).sort()?;
+        }
+        while let Some(missing) = sort::causal(&mut self.envelopes, &mut topic_cursor).sort()? {
+            let cursors = missing
+                .iter()
+                .map(|e| e.cursor())
+                .collect::<Result<Vec<_>, _>>()?;
+            // try to resolve the missing dependencies
+            let resolved = self.resolver.resolve(cursors).await?;
+            // re-apply at start of vec
+            self.envelopes.splice(0..0, resolved.into_iter());
+            sort::timestamp(&mut self.envelopes).sort()?;
+        }
+        Ok(())
+    }
+}

xmtp_api_d14n/src/protocol/resolution.rs xmtp_api_d14n/src/protocol/resolve.rsxmtp_api_d14n/src/protocol/resolution.rs renamed to xmtp_api_d14n/src/protocol/resolve.rs

@@ -7,7 +7,7 @@ pub struct CausalSort<'a, E> {
     topic_cursor: &'a mut TopicCursor,
 }
 
-impl<'a, E: Envelope<'a>> Sort<Vec<E>> for CausalSort<'a, E> {
+impl<'b, 'a: 'b, E: Envelope<'a>> Sort<Vec<E>> for CausalSort<'b, E> {
     fn sort(self) -> Result<Option<Vec<E>>, EnvelopeError> {
         let mut i = 0;
 
@@ -52,10 +52,10 @@ impl<'a, E: Envelope<'a>> Sort<Vec<E>> for CausalSort<'a, E> {
 /// * `envelopes`: the [`Envelope`]'s being sorted
 /// * `topic_cursor`: the cursor position of all known topics
 ///
-pub fn causal<'a, E: Envelope<'a>>(
-    envelopes: &'a mut Vec<E>,
-    topic_cursor: &'a mut TopicCursor,
-) -> impl Sort<Vec<E>> + use<'a, E> {
+pub fn causal<'b, 'a: 'b, E: Envelope<'a>>(
+    envelopes: &'b mut Vec<E>,
+    topic_cursor: &'b mut TopicCursor,
+) -> impl Sort<Vec<E>> + use<'a, 'b, E> {
     CausalSort {
         envelopes,
         topic_cursor,

…_d14n/src/protocol/resolution/network.rs …api_d14n/src/protocol/resolve/network.rsxmtp_api_d14n/src/protocol/resolution/network.rs renamed to xmtp_api_d14n/src/protocol/resolve/network.rs xmtp_api_d14n/src/protocol/resolution/network.rs renamed to xmtp_api_d14n/src/protocol/resolve/network.rs

@@ -4,11 +4,11 @@ pub struct TimestampSort<'a, E> {
     envelopes: &'a mut [E],
 }
 
-impl<'a, E> Sort<&'a mut [E]> for TimestampSort<'a, E>
+impl<'b, 'a: 'b, E> Sort<()> for TimestampSort<'b, E>
 where
     E: Envelope<'a>,
 {
-    fn sort(mut self) -> Result<Option<&'a mut [E]>, EnvelopeError> {
+    fn sort(mut self) -> Result<Option<()>, EnvelopeError> {
         let envelopes = &mut self.envelopes;
         // we can only sort envelopes which have a timestamp
         envelopes.sort_by_key(|e| e.timestamp());
@@ -27,9 +27,7 @@ where
 /// If a timestamp does not have a cursor (extractor return [`Option::None`]) it is
 /// sorted according to [`Ord`], [impl](https://doc.rust-lang.org/src/core/option.rs.html#2341)
 /// This sort will never return any missing envelopes.
-pub fn timestamp<'a, E: Envelope<'a>>(
-    envelopes: &'a mut [E],
-) -> impl Sort<&'a mut [E]> + use<'a, E> {
+pub fn timestamp<'b, 'a: 'b, E: Envelope<'a>>(envelopes: &'b mut [E]) -> impl Sort<()> {
     TimestampSort { envelopes }
 }
 

xmtp_api_d14n/src/protocol/sort/causal.rs

@@ -6,6 +6,7 @@ use crate::protocol::SequencedExtractor;
 use crate::protocol::V3GroupMessageExtractor;
 use crate::protocol::V3WelcomeMessageExtractor;
 use crate::protocol::WelcomeMessageExtractor;
+use derive_builder::UninitializedFieldError;
 use itertools::Itertools;
 use xmtp_proto::types::GlobalCursor;
 use xmtp_proto::types::GroupMessage;
@@ -52,6 +53,9 @@ pub use dependency_resolution::*;
 mod sort;
 pub use sort::*;
 
+mod ordered_collection;
+pub use ordered_collection::*;
+
 #[derive(thiserror::Error, Debug)]
 pub enum EnvelopeError {
     #[error(transparent)]
@@ -62,6 +66,8 @@ pub enum EnvelopeError {
     TopicMismatch,
     #[error("Envelope not found")]
     NotFound(&'static str),
+    #[error(transparent)]
+    MissingBuilderField(#[from] UninitializedFieldError),
     // for extractors defined outside of this crate or
     // generic implementations like Tuples
     #[error("{0}")]
@@ -76,6 +82,7 @@ impl RetryableError for EnvelopeError {
             Self::TopicMismatch => false,
             Self::DynError(d) => retryable!(d),
             Self::NotFound(_) => false,
+            Self::MissingBuilderField(_) => false,
         }
     }
 }

xmtp_api_d14n/src/protocol/sort/timestamp.rs

@@ -1,10 +1,12 @@
+use xmtp_common::{MaybeSend, MaybeSync};
 use xmtp_proto::types::Cursor;
 
 use crate::protocol::{Envelope, EnvelopeError};
 
-#[allow(async_fn_in_trait)]
-pub trait ResolveDependencies<'a> {
-    type ResolvedEnvelope: Envelope<'a>;
+#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
+pub trait ResolveDependencies: MaybeSend + MaybeSync {
+    type ResolvedEnvelope: Envelope<'static> + MaybeSend + MaybeSync;
     /// Resolve dependencies, starting with a list of dependencies. Should try to resolve
     /// all dependents after `dependency`, if `Dependency` is missing as well.
     /// * Once resolved, these dependencies may have missing dependencies of their own.
@@ -19,13 +21,11 @@ pub trait ResolveDependencies<'a> {
 /// A resolver that does not even attempt to try and get dependencies
 pub struct NoopResolver;
 
-#[allow(async_fn_in_trait)]
-impl ResolveDependencies<'_> for NoopResolver {
+#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
+impl ResolveDependencies for NoopResolver {
     type ResolvedEnvelope = ();
-    async fn resolve(
-        &mut self,
-        _: Vec<Cursor>,
-    ) -> Result<Vec<Self::ResolvedEnvelope>, EnvelopeError> {
+    async fn resolve(&mut self, _: Vec<Cursor>) -> Result<Vec<()>, EnvelopeError> {
         Ok(vec![])
     }
 }

xmtp_api_d14n/src/protocol/traits.rs

@@ -1,5 +1,6 @@
 //! Traits representing un-processed (extracted) and processed (extracted) protobuf types
 use chrono::Utc;
+use xmtp_common::{MaybeSend, MaybeSync};
 use xmtp_proto::types::Cursor;
 
 use crate::protocol::{CursorExtractor, DependsOnExtractor, TimestampExtractor};
@@ -13,7 +14,7 @@ use super::*;
 * Theres a way to seal this trait implementation to
 * avoid external implementations which should be done.
 */
-pub trait ProtocolEnvelope<'env>: std::fmt::Debug {
+pub trait ProtocolEnvelope<'env>: std::fmt::Debug + Send + Sync {
     type Nested<'a>
     where
         Self: 'a;
@@ -35,7 +36,7 @@ pub trait ProtocolEnvelope<'env>: std::fmt::Debug {
 /// a [`Cursor`](xmtp_proto::types::Cursor) per envelope.
 /// Likewise, Clients form the [`ClientEnvelope`] according to the [Client Node2Node Protocol](https://github.com/xmtp/XIPs/blob/main/XIPs/xip-49-decentralized-backend.md#332-envelopes)
 /// Client envelopes maintain a payload/topic with MLS and Client-specific duties.
-pub trait Envelope<'env>: std::fmt::Debug {
+pub trait Envelope<'env>: std::fmt::Debug + MaybeSend + MaybeSync {
     /// Extract the topic for this envelope
     fn topic(&self) -> Result<Topic, EnvelopeError>;
     /// Extract the cursor for this envelope

xmtp_api_d14n/src/protocol/traits/dependency_resolution.rs

@@ -0,0 +1,13 @@
+use xmtp_common::{MaybeSend, MaybeSync};
+
+use crate::protocol::EnvelopeError;
+
+// these functions are not on `EnvelopeCollection` to keep its object-safety simpler.
+// since dependency resolution requires `async fn`.
+/// A sorted envelope collection
+#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
+pub trait OrderedEnvelopeCollection: MaybeSend + MaybeSync {
+    /// Sort dependencies of `Self` according to XIP
+    async fn sort(&mut self) -> Result<(), EnvelopeError>;
+}