queries to find a dependency of an intent

.gitignore

@@ -180,3 +180,4 @@ archive.xmtp
 coverage_diff_html/**
 .direnv/**
 om.json
+*.json

Cargo.toml

@@ -151,6 +151,7 @@ wasm-bindgen-test = "0.3.51"
 web-sys = "0.3"
 web-time = "1.1"
 zeroize = "1.8"
+smallvec = "1.15"
 
 # Internal Crate Dependencies
 bindings-wasm = { path = "bindings_wasm" }

dev/test/wasm-interactive

@@ -41,5 +41,5 @@ export WASM_BINDGEN_TEST_ONLY_WEB=1
 export NO_HEADLESS=1
 
 cargo test --target wasm32-unknown-unknown --release \
-  $PACKAGE --features d14n -- \
+  -p $PACKAGE -- \
   $TESTS

xmtp_api_d14n/src/endpoints/d14n/get_newest_envelopes.rs

@@ -3,14 +3,15 @@ use prost::Message;
 use prost::bytes::Bytes;
 use std::borrow::Cow;
 use xmtp_proto::api::{BodyError, Endpoint};
+use xmtp_proto::types::Topic;
 use xmtp_proto::xmtp::xmtpv4::message_api::{GetNewestEnvelopeRequest, GetNewestEnvelopeResponse};
 
 /// Query a single thing
 #[derive(Debug, Builder, Default, Clone)]
 #[builder(build_fn(error = "BodyError"))]
 pub struct GetNewestEnvelopes {
     #[builder(setter(each(name = "topic", into)))]
-    topics: Vec<Vec<u8>>,
+    topics: Vec<Topic>,
 }
 
 impl GetNewestEnvelopes {
@@ -32,7 +33,7 @@ impl Endpoint for GetNewestEnvelopes {
 
     fn body(&self) -> Result<Bytes, BodyError> {
         let query = GetNewestEnvelopeRequest {
-            topics: self.topics.clone(),
+            topics: self.topics.iter().cloned().map(Topic::to_vec).collect(),
         };
         Ok(query.encode_to_vec().into())
     }
@@ -41,7 +42,7 @@ impl Endpoint for GetNewestEnvelopes {
 #[cfg(test)]
 mod test {
     use xmtp_api_grpc::test::XmtpdClient;
-    use xmtp_proto::{api, prelude::*};
+    use xmtp_proto::{api, prelude::*, types::TopicKind};
 
     #[xmtp_common::test]
     fn test_file_descriptor() {
@@ -67,7 +68,10 @@ mod test {
         let client = XmtpdClient::create();
         let client = client.build().unwrap();
 
-        let endpoint = GetNewestEnvelopes::builder().topic(vec![]).build().unwrap();
+        let endpoint = GetNewestEnvelopes::builder()
+            .topic(TopicKind::GroupMessagesV1.create(vec![]))
+            .build()
+            .unwrap();
         api::ignore(endpoint).query(&client).await.unwrap();
     }
 }

xmtp_api_d14n/src/endpoints/d14n/publish_client_envelopes.rs

@@ -74,7 +74,7 @@ mod test {
         let client = client.build().unwrap();
 
         let aad = AuthenticatedData {
-            target_topic: TopicKind::GroupMessagesV1.build(rand_vec::<16>()),
+            target_topic: TopicKind::GroupMessagesV1.create(rand_vec::<16>()).to_vec(),
             depends_on: None,
         };
         let e = ClientEnvelope {

xmtp_api_d14n/src/endpoints/d14n/query_envelopes.rs

@@ -32,7 +32,7 @@ impl Endpoint for QueryEnvelope {
     fn body(&self) -> Result<Bytes, BodyError> {
         let query = QueryEnvelopesRequest {
             query: Some(EnvelopesQuery {
-                topics: self.topics.iter().map(Topic::bytes).collect(),
+                topics: self.topics.iter().cloned().map(Topic::to_vec).collect(),
                 originator_node_ids: vec![],
                 last_seen: Some(self.last_seen.clone().into()),
             }),

xmtp_api_d14n/src/endpoints/d14n/subscribe_envelopes.rs

@@ -36,7 +36,7 @@ impl Endpoint for SubscribeEnvelopes {
             tracing::info!("subscribing to {}", topic.clone());
         }
         let query = EnvelopesQuery {
-            topics: self.topics.iter().map(Topic::bytes).collect(),
+            topics: self.topics.iter().cloned().map(Topic::to_vec).collect(),
             last_seen: self.last_seen.clone().map(Into::into),
             originator_node_ids: self.originators.clone(),
         };

xmtp_api_d14n/src/middleware/multi_node_client/client.rs

@@ -84,10 +84,10 @@ mod tests {
     };
     use std::sync::Arc;
     use xmtp_configuration::{GrpcUrls, PAYER_WRITE_FILTER};
-    use xmtp_proto::api::Query;
     use xmtp_proto::api_client::{ApiBuilder, NetConnectConfig};
     use xmtp_proto::prelude::XmtpMlsClient;
     use xmtp_proto::types::GroupId;
+    use xmtp_proto::{api::Query, types::TopicKind};
 
     fn is_tls_enabled() -> bool {
         url::Url::parse(GrpcUrls::GATEWAY)
@@ -273,7 +273,10 @@ mod tests {
     async fn multinode_request_latest_group_message() {
         use crate::d14n::GetNewestEnvelopes;
         let client = create_multinode_client();
-        let mut endpoint = GetNewestEnvelopes::builder().topic(vec![]).build().unwrap();
+        let mut endpoint = GetNewestEnvelopes::builder()
+            .topic(TopicKind::GroupMessagesV1.create(vec![]))
+            .build()
+            .unwrap();
         let response = endpoint.query(&client).await.unwrap();
         assert!(!response.results.is_empty());
     }

xmtp_api_d14n/src/protocol/extractors/data.rs

@@ -1,5 +1,6 @@
 //! Extractor for an MLS Data field
 //! useful for verifing a message has been read or maybe duplicates.
+use xmtp_cryptography::hash::sha256_bytes;
 use xmtp_proto::ConversionError;
 use xmtp_proto::mls_v1::group_message_input::V1 as GroupMessageV1;
 use xmtp_proto::mls_v1::welcome_message_input::V1 as WelcomeMessageV1;
@@ -19,7 +20,12 @@ impl MlsDataExtractor {
     pub fn new() -> Self {
         Default::default()
     }
+
+    pub fn as_sha256(&self) -> Option<Vec<u8>> {
+        self.data.as_ref().map(|d| sha256_bytes(d))
+    }
 }
+
 impl Extractor for MlsDataExtractor {
     type Output = Result<Vec<u8>, ConversionError>;
 

xmtp_api_d14n/src/protocol/in_memory_cursor_store.rs

@@ -3,6 +3,8 @@ use std::collections::HashMap;
 use std::fmt;
 use xmtp_proto::types::{GlobalCursor, OriginatorId, Topic};
 
+// an in memory cursor store that cannot track dependencies
+// should only be used for tests
 #[derive(Default, Clone)]
 pub struct InMemoryCursorStore {
     topics: HashMap<Topic, GlobalCursor>,
@@ -185,7 +187,7 @@ mod tests {
     }
 
     fn topic(name: &str) -> Topic {
-        Topic::from_bytes(name.as_bytes().to_vec())
+        Topic::from_bytes_unchecked(name.as_bytes().to_vec())
     }
 
     #[xmtp_common::test]

xmtp_api_d14n/src/protocol/sort/causal.rs

@@ -1,6 +1,6 @@
 use xmtp_proto::types::TopicCursor;
 
-use crate::protocol::{ApplyCursor, Envelope, EnvelopeError, Sort, VectorClock};
+use crate::protocol::{ApplyEnvelope, Envelope, EnvelopeError, Sort, VectorClock};
 
 pub struct CausalSort<'a, E> {
     envelopes: &'a mut Vec<E>,
@@ -18,7 +18,7 @@ impl<'b, 'a: 'b, E: Envelope<'a>> Sort<Vec<E>> for CausalSort<'b, E> {
             let last_seen = env.depends_on()?.unwrap_or(Default::default());
             let vector_clock = self.topic_cursor.get_or_default(&topic);
             if vector_clock.dominates(&last_seen) {
-                self.topic_cursor.apply(env)?;
+                self.topic_cursor.apply_envelope(env)?;
                 i += 1;
             } else {
                 missing.push(self.envelopes.remove(i));

xmtp_api_d14n/src/protocol/traits/sort.rs

@@ -1,4 +1,4 @@
-use xmtp_proto::types::{GlobalCursor, TopicCursor};
+use xmtp_proto::types::TopicCursor;
 
 use crate::protocol::{Envelope, EnvelopeError};
 
@@ -19,24 +19,16 @@ pub trait Sort<Missing> {
 
 /// Extension trait to modify a [`TopicCursor`]
 /// with the contents of an envelope.
-pub trait ApplyCursor<E> {
+pub trait ApplyEnvelope<E> {
     /// applies an envelope to a cursor
-    fn apply(&mut self, envelope: &E) -> Result<(), EnvelopeError>;
+    fn apply_envelope(&mut self, envelope: &E) -> Result<(), EnvelopeError>;
 }
 
-impl<'a, E: Envelope<'a>> ApplyCursor<E> for TopicCursor {
-    fn apply(&mut self, envelope: &E) -> Result<(), EnvelopeError> {
+impl<'a, E: Envelope<'a>> ApplyEnvelope<E> for TopicCursor {
+    fn apply_envelope(&mut self, envelope: &E) -> Result<(), EnvelopeError> {
         let topic = envelope.topic()?;
         let cursor = envelope.cursor()?;
-        self.entry(topic)
-            .and_modify(|global| {
-                global.apply(&cursor);
-            })
-            .or_insert_with(|| {
-                let mut map = GlobalCursor::default();
-                map.apply(&cursor);
-                map
-            });
+        self.apply(topic, &cursor);
         Ok(())
     }
 }

xmtp_api_d14n/src/queries/d14n/identity.rs

@@ -77,7 +77,7 @@ where
         });
         let result: QueryEnvelopesResponse = QueryEnvelopes::builder()
             .envelopes(EnvelopesQuery {
-                topics: topics.iter().map(Topic::bytes).collect(),
+                topics: topics.iter().cloned().map(Topic::to_vec).collect(),
                 originator_node_ids: vec![],
                 last_seen,
             })

xmtp_api_d14n/src/queries/d14n/mls.rs

@@ -18,8 +18,8 @@ use crate::protocol::traits::Extractor;
 use crate::queries::D14nCombinatorExt;
 use xmtp_common::RetryableError;
 use xmtp_configuration::MAX_PAGE_SIZE;
-use xmtp_proto::api;
 use xmtp_proto::api::Client;
+use xmtp_proto::api::EndpointExt;
 use xmtp_proto::api::{ApiClientError, Query};
 use xmtp_proto::api_client::XmtpMlsClient;
 use xmtp_proto::mls_v1;
@@ -49,13 +49,12 @@ where
         request: mls_v1::UploadKeyPackageRequest,
     ) -> Result<(), Self::Error> {
         let envelopes = request.client_envelope()?;
-        api::ignore(
-            PublishClientEnvelopes::builder()
-                .envelope(envelopes)
-                .build()?,
-        )
-        .query(&self.client)
-        .await?;
+        PublishClientEnvelopes::builder()
+            .envelope(envelopes)
+            .build()?
+            .ignore_response()
+            .query(&self.client)
+            .await?;
 
         Ok::<_, Self::Error>(())
     }
@@ -68,7 +67,7 @@ where
         let topics = request
             .installation_keys
             .iter()
-            .map(|key| TopicKind::KeyPackagesV1.build(key))
+            .map(|key| TopicKind::KeyPackagesV1.create(key))
             .collect();
 
         let result: GetNewestEnvelopeResponse = GetNewestEnvelopes::builder()
@@ -89,13 +88,12 @@ where
     ) -> Result<(), Self::Error> {
         let envelopes: Vec<ClientEnvelope> = request.messages.client_envelopes()?;
 
-        api::ignore(
-            PublishClientEnvelopes::builder()
-                .envelopes(envelopes)
-                .build()?,
-        )
-        .query(&self.client)
-        .await?;
+        PublishClientEnvelopes::builder()
+            .envelopes(envelopes)
+            .build()?
+            .ignore_response()
+            .query(&self.client)
+            .await?;
 
         Ok(())
     }
@@ -107,13 +105,12 @@ where
     ) -> Result<(), Self::Error> {
         let envelopes = request.messages.client_envelopes()?;
 
-        api::ignore(
-            PublishClientEnvelopes::builder()
-                .envelopes(envelopes)
-                .build()?,
-        )
-        .query(&self.client)
-        .await?;
+        PublishClientEnvelopes::builder()
+            .envelopes(envelopes)
+            .build()?
+            .ignore_response()
+            .query(&self.client)
+            .await?;
         Ok(())
     }
 
@@ -152,7 +149,7 @@ where
         group_id: GroupId,
     ) -> Result<Option<xmtp_proto::types::GroupMessage>, Self::Error> {
         let response: GetNewestEnvelopeResponse = GetNewestEnvelopes::builder()
-            .topic(TopicKind::GroupMessagesV1.build(group_id))
+            .topic(TopicKind::GroupMessagesV1.create(group_id))
             .build()?
             .query(&self.client)
             .await?;
@@ -216,10 +213,10 @@ where
         &self,
         request: mls_v1::GetNewestGroupMessageRequest,
     ) -> Result<Vec<Option<GroupMessageMetadata>>, Self::Error> {
-        let topics: Vec<Vec<u8>> = request
+        let topics = request
             .group_ids
             .into_iter()
-            .map(|id| TopicKind::GroupMessagesV1.build(id.as_slice()))
+            .map(|id| TopicKind::GroupMessagesV1.create(id))
             .collect();
 
         let response = GetNewestEnvelopes::builder()

xmtp_api_d14n/src/queries/v3/xmtp_query.rs

@@ -84,7 +84,6 @@ where
                     .await?;
                 Ok(XmtpEnvelope::new(result.key_packages))
             }
-            _ => unreachable!(),
         }
     }
 }

xmtp_db/Cargo.toml

@@ -79,6 +79,7 @@ wasm-bindgen = { workspace = true }
 
 
 [dev-dependencies]
+futures.workspace = true
 futures-timer.workspace = true
 mockall = { workspace = true }
 rstest.workspace = true
@@ -94,9 +95,17 @@ tokio = { workspace = true, features = [
   "rt",
   "rt-multi-thread",
 ] }
+proptest = { workspace = true, features = [
+  "std",
+  "fork",
+  "timeout",
+  "bit-set",
+] }
+
 
 [target.'cfg(target_arch = "wasm32")'.dev-dependencies]
 wasm-bindgen-test.workspace = true
+proptest = { workspace = true, features = ["std"] }
 
 [features]
 update-schema = ["dep:toml"]

xmtp_db/proptest-regressions/encrypted_store/group_intent.txt

@@ -0,0 +1,9 @@
+# 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 b2215054455cbc58813defa02e820702950c23e383568a2839cc2c351a313c2a # shrinks to intents_data = [IntentWithDependency { payload_hash: [107, 102, 193, 176, 236, 243, 34, 27, 30, 93, 100, 228, 255, 37, 147, 238, 69, 34, 25, 241, 78, 213, 61, 4], intent_epoch: 1, dependency: Some((0, 0)) }]
+cc a45ea5ca4aadf0c893410edd4b0ee7e0d98e1f5a67674941153431e2398fffe3 # shrinks to intents_data = [IntentWithDependency { payload_hash: [32, 167, 202, 237, 18, 189, 112, 135, 50, 124, 240, 141, 171, 159, 103, 238, 34, 196, 205, 5, 111, 38, 54, 126], intent_epoch: 22, dependency: None }, IntentWithDependency { payload_hash: [36, 11, 54, 242, 159, 139, 230, 170, 120, 26, 127, 151, 60, 107, 106, 248, 209, 24, 50, 255, 134, 13, 181, 29], intent_epoch: 22, dependency: Some((0, 0)) }]
+cc 3314fd486372023c80bf55e95d246a661b5673e6800f7ba7da1b257371f90179 # shrinks to intents_data = [IntentWithDependency { payload_hash: 7d42b35b133d915e0456eca9b1d8cdb15f9b297fef9843cc, intent_epoch: 1, dependency: None }, IntentWithDependency { payload_hash: 4dd10ffa10fa332e0c956f13e17dabdfe10c5b204561bf5b, intent_epoch: 1, dependency: None }, IntentWithDependency { payload_hash: b2f82c6e70e86302babd69372bb0631e51e4d5efb01c7d89, intent_epoch: 1, dependency: None }, IntentWithDependency { payload_hash: 7bc46b4d349c1667ed5ad051851ece808a45628b9e95b7a1, intent_epoch: 1, dependency: None }, IntentWithDependency { payload_hash: 758ea93d88903298dd0a5e8ef4ed11d62b0b18940c528f83, intent_epoch: 1, dependency: None }, IntentWithDependency { payload_hash: 6130a674328894f12592cc27a736f85c801e5ad31ccd1b16, intent_epoch: 1, dependency: None }, IntentWithDependency { payload_hash: a31cfc7a2f0e62e09eea4e6f9d8af0a0f8e9d214b56181b0, intent_epoch: 66, dependency: Some((0, 0)) }, IntentWithDependency { payload_hash: fe8263711c669f597e3be37df33dae706b5e4b0588a299cb, intent_epoch: 66, dependency: Some((0, 0)) }, IntentWithDependency { payload_hash: 05e3a30b2372c88950eb192dd25dc96f15b0d2583b759b63, intent_epoch: 1, dependency: None }, IntentWithDependency { payload_hash: 675568a6939ddbdb667ed75267d23fc98b620513ce182529, intent_epoch: 1, dependency: None }]

xmtp_db/proptest-regressions/encrypted_store/refresh_state.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 e4e304439a2b33516530ab44b01a030988f41aa1be25e6c8a24f401b10b30d3c # shrinks to topics = [Topic { kind: GroupMessagesV1, bytes: "00" }], states = [(210, 0), (210, 1)]