add resolution strategy

Author: insipx

xmtp_api_d14n/src/protocol/mod.rs

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

xmtp_api_d14n/src/protocol/order.rs

@@ -0,0 +1,52 @@
+use crate::protocol::{
+    Envelope, EnvelopeError, OrderedEnvelopeCollection, ResolveDependencies, Sort, sort,
+};
+use derive_builder::Builder;
+use xmtp_proto::types::TopicCursor;
+
+/// Order dependencies of `Self` according to [XIP](https://github.com/xmtp/XIPs/blob/main/XIPs/xip-49-decentralized-backend.md#335-cross-originator-message-ordering)
+/// If dependencies are missing, this ordering will try to resolve them
+/// and re-apply resolved dependencies to the front of the envelope list
+/// construct this strategy with [`Ordered::builder`]
+#[derive(Debug, Clone, Builder)]
+#[builder(setter(strip_option), build_fn(error = "EnvelopeError"))]
+pub struct Ordered<T, R> {
+    envelopes: Vec<T>,
+    resolver: R,
+    topic_cursor: TopicCursor,
+}
+
+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>,
+{
+    async fn order(&mut self) -> Result<(), EnvelopeError> {
+        let Self {
+            envelopes,
+            resolver,
+            topic_cursor,
+        } = self;
+        sort::timestamp(envelopes).sort()?;
+        while let Some(missing) = sort::causal(envelopes, topic_cursor).sort()? {
+            let cursors = missing
+                .iter()
+                .map(|e| e.cursor())
+                .collect::<Result<Vec<_>, _>>()?;
+            // try to resolve the missing dependencies
+            let resolved = resolver.resolve(cursors).await?;
+            // re-apply at start of vec
+            envelopes.splice(0..0, resolved.into_iter());
+            sort::timestamp(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

@@ -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,
         }
     }
 }

…_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

@@ -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.rs

@@ -0,0 +1,17 @@
+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 ordered envelope collection
+/// an `OrderedEnvelopeCollection` differs from [`Sort`](super::Sort)
+/// since it adds the inclusing of `async`, allowing
+/// an `OrderedEnvelopeCollection`  to both
+/// [Sort](super::Sort) and [ResolveDependencies](super::ResolveDependencies)
+#[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 {
+    /// Order dependencies of `Self` according to [XIP](https://github.com/xmtp/XIPs/blob/main/XIPs/xip-49-decentralized-backend.md#335-cross-originator-message-ordering)
+    async fn order(&mut self) -> Result<(), EnvelopeError>;
+}

xmtp_api_d14n/src/protocol/traits/envelopes.rs

@@ -6,7 +6,7 @@ use std::{
 use crate::types::{GlobalCursor, Topic};
 
 /// A cursor that keeps a [`super::GlobalCursor`] for each topic it has seen.
-#[derive(Default, PartialEq)]
+#[derive(Default, Debug, PartialEq, Clone)]
 pub struct TopicCursor {
     inner: HashMap<Topic, GlobalCursor>,
 }