simple timestamp sort

Cargo.toml

@@ -108,6 +108,7 @@ parking_lot = "0.12.3"
 pbjson = "0.8"
 pbjson-types = "0.8"
 pin-project-lite = "0.2"
+proptest = { version = "1.9", default-features = false }
 prost = { version = "0.14", default-features = false }
 prost-types = { version = "0.14", default-features = false }
 # updating crates are blocked on https://github.com/dalek-cryptography/curve25519-dalek/pull/729

xmtp_api_d14n/Cargo.toml

@@ -56,11 +56,18 @@ xmtp_proto = { workspace = true, features = ["test-utils"] }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
 ctor.workspace = true
+proptest = { workspace = true, features = [
+  "std",
+  "fork",
+  "timeout",
+  "bit-set",
+] }
 
 [target.'cfg(target_arch = "wasm32")'.dev-dependencies]
 wasm-bindgen-test.workspace = true
 # needed for rstest
 futures-timer = { workspace = true, features = ["wasm-bindgen"] }
+proptest = { workspace = true, features = ["std"] }
 
 
 [features]

xmtp_api_d14n/proptest-regressions/protocol/sort/timestamp.txt

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 783ebaf4bd78f8d0bb95fa2dbd7c798aee6abfb3d504dbc7ac1f7724fd1bb326 # shrinks to mut envelopes = [TestEnvelope { time: Some(0) }, TestEnvelope { time: Some(-1) }]

xmtp_api_d14n/src/protocol/extractors.rs

@@ -21,6 +21,10 @@ mod topics;
 pub use topics::*;
 mod data;
 pub use data::*;
+mod cursor;
+pub use cursor::*;
+mod timestamp;
+pub use timestamp::*;
 
 #[cfg(test)]
 pub mod test_utils;

xmtp_api_d14n/src/protocol/extractors/cursor.rs

@@ -0,0 +1,64 @@
+//! Extractor for a envelope [`Cursor`](xmtp_proto::types::Cursor)
+//! useful for verifing a message has been read or maybe duplicates.
+use xmtp_proto::ConversionError;
+use xmtp_proto::mls_v1::welcome_message::WelcomePointer as V3WelcomePointer;
+use xmtp_proto::types::Cursor;
+use xmtp_proto::xmtp::mls::api::v1::{
+    group_message::V1 as V3GroupMessage, welcome_message::V1 as V3WelcomeMessage,
+};
+use xmtp_proto::xmtp::xmtpv4::envelopes::UnsignedOriginatorEnvelope;
+
+use crate::protocol::{EnvelopeVisitor, Extractor};
+
+/// Extract Cursor from Envelopes
+#[derive(Default, Clone, Debug)]
+pub struct CursorExtractor {
+    cursor: Option<Cursor>,
+}
+
+impl CursorExtractor {
+    pub fn new() -> Self {
+        Default::default()
+    }
+}
+
+impl Extractor for CursorExtractor {
+    type Output = Result<Cursor, ConversionError>;
+
+    fn get(self) -> Self::Output {
+        self.cursor.ok_or(ConversionError::Missing {
+            item: "cursor",
+            r#type: std::any::type_name::<Cursor>(),
+        })
+    }
+}
+
+impl EnvelopeVisitor<'_> for CursorExtractor {
+    type Error = ConversionError;
+
+    fn visit_unsigned_originator(
+        &mut self,
+        e: &UnsignedOriginatorEnvelope,
+    ) -> Result<(), Self::Error> {
+        self.cursor = Some(Cursor {
+            sequence_id: e.originator_sequence_id,
+            originator_id: e.originator_node_id,
+        });
+        Ok(())
+    }
+
+    fn visit_v3_group_message(&mut self, m: &V3GroupMessage) -> Result<(), Self::Error> {
+        self.cursor = Some(Cursor::v3_messages(m.id));
+        Ok(())
+    }
+
+    fn visit_v3_welcome_message(&mut self, m: &V3WelcomeMessage) -> Result<(), Self::Error> {
+        self.cursor = Some(Cursor::v3_welcomes(m.id));
+        Ok(())
+    }
+
+    fn visit_v3_welcome_pointer(&mut self, m: &V3WelcomePointer) -> Result<(), Self::Error> {
+        self.cursor = Some(Cursor::v3_welcomes(m.id));
+        Ok(())
+    }
+}

xmtp_api_d14n/src/protocol/extractors/timestamp.rs

@@ -0,0 +1,57 @@
+//! Extractor for an MLS Data field
+//! useful for verifing a message has been read or maybe duplicates.
+use chrono::Utc;
+use xmtp_proto::ConversionError;
+use xmtp_proto::mls_v1::welcome_message::WelcomePointer;
+use xmtp_proto::xmtp::mls::api::v1::{
+    group_message::V1 as V3GroupMessage, welcome_message::V1 as V3WelcomeMessage,
+};
+use xmtp_proto::xmtp::xmtpv4::envelopes::UnsignedOriginatorEnvelope;
+
+use crate::protocol::{EnvelopeVisitor, Extractor};
+
+/// Extract Mls Data from Envelopes
+#[derive(Default, Clone, Debug)]
+pub struct TimestampExtractor {
+    time: Option<i64>,
+}
+
+impl TimestampExtractor {
+    pub fn new() -> Self {
+        Default::default()
+    }
+}
+
+impl Extractor for TimestampExtractor {
+    type Output = Option<chrono::DateTime<Utc>>;
+
+    fn get(self) -> Self::Output {
+        self.time.map(chrono::DateTime::from_timestamp_nanos)
+    }
+}
+
+impl EnvelopeVisitor<'_> for TimestampExtractor {
+    type Error = ConversionError;
+
+    fn visit_unsigned_originator(
+        &mut self,
+        e: &UnsignedOriginatorEnvelope,
+    ) -> Result<(), Self::Error> {
+        self.time = Some(e.originator_ns);
+        Ok(())
+    }
+
+    fn visit_v3_group_message(&mut self, message: &V3GroupMessage) -> Result<(), Self::Error> {
+        self.time = Some(message.created_ns as i64);
+        Ok(())
+    }
+
+    fn visit_v3_welcome_message(&mut self, message: &V3WelcomeMessage) -> Result<(), Self::Error> {
+        self.time = Some(message.created_ns as i64);
+        Ok(())
+    }
+    fn visit_v3_welcome_pointer(&mut self, ptr: &WelcomePointer) -> Result<(), Self::Error> {
+        self.time = Some(ptr.created_ns as i64);
+        Ok(())
+    }
+}

xmtp_api_d14n/src/protocol/mod.rs

@@ -10,3 +10,10 @@ mod in_memory_cursor_store;
 pub use in_memory_cursor_store::*;
 
 mod impls;
+
+mod resolution;
+// pub use resolution::*;
+
+pub mod sort;
+
+pub mod types;

xmtp_api_d14n/src/protocol/resolution.rs

@@ -0,0 +1,10 @@
+//! Implementations for Dependency Resolution strategies [XIP](https://github.com/xmtp/XIPs/blob/main/XIPs/xip-49-decentralized-backend.md#335-cross-originator-message-ordering)
+//!
+//! Possible Implementation of Dependency Resolution Strategies:
+//! - keep retrying same query and error forever after and finish after some backoff
+//! - query the originator that the mesasge is stored on.
+//! - file misbehavior report if originator message came from is unresponsive
+//! - dont resolve dependencies at all
+//! - query random originators for the dependency
+//! - round robin query for dependency
+//! - etc.

xmtp_api_d14n/src/protocol/sort.rs

@@ -0,0 +1,5 @@
+//! Sorting Implelementations on Envelope Collections [XIP](https://github.com/xmtp/XIPs/blob/main/XIPs/xip-49-decentralized-backend.md#335-cross-originator-message-ordering)
+// mod causal;
+// pub use causal::*;
+mod timestamp;
+pub use timestamp::*;

xmtp_api_d14n/src/protocol/sort/timestamp.rs

@@ -0,0 +1,125 @@
+use crate::protocol::{Envelope, Sort};
+
+pub struct TimestampSort<'a, E> {
+    envelopes: &'a mut [E],
+}
+
+impl<'b, 'a, E> Sort for TimestampSort<'b, E>
+where
+    E: Envelope<'a>,
+{
+    fn sort(mut self) {
+        let envelopes = &mut self.envelopes;
+        // we can only sort envelopes which have a timestamp
+        envelopes.sort_unstable_by_key(|e| e.timestamp());
+    }
+}
+
+/// Sorts Envelopes by server-side Timestamp in ascending order
+/// * for d14n this will sort envelopes by
+///   [`originator_ns`](xmtp_proto::xmtp::xmtpv4::envelopes::UnsignedOriginatorEnvelope::originator_ns)
+/// * for v3 this will sort by created_ns on GroupMessage, WelcomeMessage, or WelcomePointer
+///   overall, sorts according to the timestamp extracted by
+///   [`TimestampExtractor`](crate::protocol::TimestampExtractor)
+///
+/// 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<'b, 'a: 'b, E: Envelope<'a>>(envelopes: &'b mut [E]) -> impl Sort {
+    TimestampSort { envelopes }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::protocol::sort;
+    use chrono::Utc;
+    use proptest::prelude::*;
+    use xmtp_common::Generate;
+    use xmtp_proto::xmtp::xmtpv4::envelopes::ClientEnvelope;
+    use xmtp_proto::xmtp::xmtpv4::envelopes::client_envelope::Payload;
+
+    use super::*;
+
+    #[derive(Debug)]
+    struct TestEnvelope {
+        time: Option<chrono::DateTime<Utc>>,
+    }
+
+    impl TestEnvelope {
+        fn new(time: i64) -> Self {
+            Self {
+                time: Some(chrono::DateTime::from_timestamp_nanos(time)),
+            }
+        }
+    }
+
+    impl Generate for TestEnvelope {
+        fn generate() -> Self {
+            TestEnvelope {
+                time: Some(chrono::DateTime::from_timestamp_nanos(
+                    xmtp_common::rand_i64(),
+                )),
+            }
+        }
+    }
+
+    impl Envelope<'_> for TestEnvelope {
+        fn topic(&self) -> Result<xmtp_proto::types::Topic, crate::protocol::EnvelopeError> {
+            unreachable!()
+        }
+
+        fn payload(&self) -> Result<Payload, crate::protocol::EnvelopeError> {
+            unreachable!()
+        }
+
+        fn timestamp(&self) -> Option<chrono::DateTime<Utc>> {
+            self.time
+        }
+
+        fn client_envelope(&self) -> Result<ClientEnvelope, crate::protocol::EnvelopeError> {
+            unreachable!()
+        }
+
+        fn group_message(
+            &self,
+        ) -> Result<Option<xmtp_proto::types::GroupMessage>, crate::protocol::EnvelopeError>
+        {
+            unreachable!()
+        }
+
+        fn welcome_message(
+            &self,
+        ) -> Result<Option<xmtp_proto::types::WelcomeMessage>, crate::protocol::EnvelopeError>
+        {
+            unreachable!()
+        }
+
+        fn consume<E>(&self, _extractor: E) -> Result<E::Output, crate::protocol::EnvelopeError>
+        where
+            Self: Sized,
+            for<'a> crate::protocol::EnvelopeError:
+                From<<E as crate::protocol::EnvelopeVisitor<'a>>::Error>,
+            for<'a> E: crate::protocol::EnvelopeVisitor<'a> + crate::protocol::Extractor,
+        {
+            unreachable!()
+        }
+    }
+
+    prop_compose! {
+        fn envelope_all_some()(id in 1..u32::MAX) -> TestEnvelope {
+            TestEnvelope::new(id as i64)
+        }
+    }
+
+    fn is_sorted(sorted: &[TestEnvelope]) -> bool {
+        sorted.is_sorted_by_key(|e| e.time)
+    }
+
+    #[xmtp_common::test]
+    fn sorts_by_timestamp() {
+        proptest!(|(mut envelopes in prop::collection::vec(envelope_all_some(), 0 .. 100))| {
+            sort::timestamp(&mut envelopes).sort();
+            assert!(is_sorted(&envelopes), "envelopes were not sorted")
+        });
+    }
+}

xmtp_api_d14n/src/protocol/traits.rs

@@ -46,6 +46,12 @@ pub use vector_clock::*;
 mod full_api;
 pub use full_api::*;
 
+mod dependency_resolution;
+pub use dependency_resolution::*;
+
+mod sort;
+pub use sort::*;
+
 #[derive(thiserror::Error, Debug)]
 pub enum EnvelopeError {
     #[error(transparent)]

xmtp_api_d14n/src/protocol/traits/dependency_resolution.rs

@@ -0,0 +1,30 @@
+use xmtp_common::{MaybeSend, MaybeSync};
+
+use crate::protocol::{Envelope, EnvelopeError, types::MissingEnvelope};
+
+#[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.
+    /// # Returns
+    /// * `Vec<Self::ResolvedEnvelope>`: The list of envelopes which were resolved.
+    async fn resolve(
+        &mut self,
+        missing: Vec<MissingEnvelope>,
+    ) -> Result<Vec<Self::ResolvedEnvelope>, EnvelopeError>;
+}
+
+/// A resolver that does not even attempt to try and get dependencies
+pub struct 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<MissingEnvelope>) -> Result<Vec<()>, EnvelopeError> {
+        Ok(vec![])
+    }
+}