chore: wip

openmls/Cargo.toml

@@ -38,6 +38,7 @@ openmls_test = { path = "../openmls_test", optional = true }
 openmls_libcrux_crypto = { path = "../libcrux_crypto", optional = true }
 maybe-async = {workspace = true}
 async-trait = { workspace = true, optional = true }
+futures = { version = "0.3.30", optional = true }
 
 [features]
 default = ["backtrace"]
@@ -62,7 +63,7 @@ js = [
     "dep:getrandom",
     "dep:fluvio-wasm-timer",
 ] # enable js randomness source for provider
-async = ["dep:async-trait", "openmls_traits/async"]
+async = ["dep:async-trait", "openmls_traits/async", "dep:futures"]
 
 [dev-dependencies]
 backtrace = "0.3"

openmls/src/group/core_group/process.rs

@@ -1,3 +1,6 @@
+#[cfg(feature = "async")]
+use futures::{stream, StreamExt};
+use futures::TryStreamExt;
 use core_group::proposals::QueuedProposal;
 
 use crate::{
@@ -280,15 +283,34 @@ impl CoreGroup {
         // If we are processing an update proposal that originally came from
         // us, the keypair corresponding to the leaf in the update is also a
         // potential decryption keypair.
-        let leaf_node_keypairs = own_leaf_nodes
+        let leaf_node_keypairs = Self::encryption_key_pairs_from_own_leaf_nodes(provider, own_leaf_nodes).await;
+
+        Ok((old_epoch_keypairs, leaf_node_keypairs))
+    }
+
+    #[cfg(feature = "async")]
+    async fn encryption_key_pairs_from_own_leaf_nodes(provider: &impl OpenMlsProvider, own_leaf_nodes: &[LeafNode]) -> Vec<EncryptionKeyPair> {
+        let stream = stream::iter(own_leaf_nodes);
+        let then = stream.then(|leaf_node| async {
+                EncryptionKeyPair::read(provider, leaf_node.encryption_key()).await
+                    .ok_or(StageCommitError::MissingDecryptionKey)
+            });
+        then.map(|e| {
+            e.and_then(|e| {
+                Ok(e)
+            })
+        }).collect::<Vec<_>>()
+        }
+
+    #[cfg(not(feature = "async"))]
+    async fn encryption_key_pairs_from_own_leaf_nodes(provider: &impl OpenMlsProvider, own_leaf_nodes: &[LeafNode]) -> Vec<EncryptionKeyPair> {
+        own_leaf_nodes
             .iter()
             .map(|leaf_node| {
                 EncryptionKeyPair::read(provider, leaf_node.encryption_key())
                     .ok_or(StageCommitError::MissingDecryptionKey)
             })
-            .collect::<Result<Vec<EncryptionKeyPair>, StageCommitError>>()?;
-
-        Ok((old_epoch_keypairs, leaf_node_keypairs))
+            .collect::<Result<Vec<EncryptionKeyPair>, StageCommitError>>()
     }
 
     /// Merge a [StagedCommit] into the group after inspection

openmls/src/group/mls_group/creation.rs

@@ -1,4 +1,6 @@
 use openmls_traits::{signatures::Signer, storage::StorageProvider as StorageProviderTrait};
+#[cfg(feature = "async")]
+use futures::{stream::{self, StreamExt}, TryFutureExt};
 
 use super::{builder::MlsGroupBuilder, *};
 use crate::{
@@ -164,7 +166,7 @@ impl ProcessedWelcome {
         welcome: Welcome,
     ) -> Result<Self, WelcomeError<Provider::StorageError>> {
         let (resumption_psk_store, key_package_bundle) =
-            keys_for_welcome(mls_group_config, &welcome, provider).await.await?;
+            keys_for_welcome(mls_group_config, &welcome, provider).await?;
 
         let (ciphersuite, group_secrets, key_schedule, verifiable_group_info) =
             crate::group::core_group::new_from_welcome::process_welcome(
@@ -316,29 +318,47 @@ async fn keys_for_welcome<Provider: OpenMlsProvider>(
     WelcomeError<<Provider as OpenMlsProvider>::StorageError>,
 > {
     let resumption_psk_store = ResumptionPskStore::new(mls_group_config.number_of_resumption_psks);
-    let key_package_bundle: KeyPackageBundle = {
-        welcome
-            .secrets()
-            .iter()
-            .find_map(|egs| {
-                let hash_ref = egs.new_member();
-
-                transpose_err_opt(
-                    provider
-                        .storage()
-                        .key_package(&hash_ref)
-                        .map_err(WelcomeError::StorageError),
-                )
-            })
-            .ok_or(WelcomeError::NoMatchingKeyPackage)??
-    };
+    let key_package_bundle = get_key_package_bundle_for_welcome(welcome, provider).await.ok_or(WelcomeError::NoMatchingKeyPackage)?;
     if !key_package_bundle.key_package().last_resort() {
         provider
             .storage()
-            .delete_key_package(&key_package_bundle.key_package.hash_ref(provider.crypto())?)
+            .delete_key_package(&key_package_bundle.key_package.hash_ref(provider.crypto())?).await
             .map_err(WelcomeError::StorageError)?;
     } else {
         log::debug!("Key package has last resort extension, not deleting");
     }
     Ok((resumption_psk_store, key_package_bundle))
 }
+
+#[maybe_async::must_be_async]
+#[cfg(feature = "async")]
+async fn get_key_package_bundle_for_welcome<Provider: OpenMlsProvider>(welcome: &Welcome, provider: &Provider) -> Option<KeyPackageBundle> {
+        let stream = stream::iter(welcome.secrets());
+        let events = stream.filter_map(|egs| async move {
+            let hash_ref = egs.new_member();
+            provider
+                .storage()
+                .key_package(&hash_ref).await
+                .ok()?
+        });
+        events.collect::<Vec<KeyPackageBundle>>().await.first().cloned()
+}
+
+
+#[maybe_async::must_be_sync]
+#[cfg(not(feature = "async"))]
+async fn get_key_package_bundle_for_welcome<Provider: OpenMlsProvider>(welcome: &Welcome, provider: &Provider) ->  Option<KeyPackageBundle> {
+    welcome
+        .secrets()
+        .iter()
+        .find_map(|egs| {
+            let hash_ref = egs.new_member();
+
+            transpose_err_opt(
+                provider
+                    .storage()
+                    .key_package(&hash_ref).await
+                    .map_err(WelcomeError::StorageError),
+            )
+        })?.ok()
+}

openmls/src/group/mls_group/updates.rs

@@ -5,6 +5,8 @@ use crate::{messages::group_info::GroupInfo, storage::OpenMlsProvider, treesync:
 
 use super::*;
 
+#[cfg_attr(feature = "async", maybe_async::must_be_async)]
+#[cfg_attr(not(feature = "async"), maybe_async::must_be_sync)]
 impl MlsGroup {
     /// Updates the own leaf node.
     ///
@@ -23,7 +25,7 @@ impl MlsGroup {
     /// [`Welcome`]: crate::messages::Welcome
     // FIXME: #1217
     #[allow(clippy::type_complexity)]
-    pub fn self_update<Provider: OpenMlsProvider>(
+    pub async fn self_update<Provider: OpenMlsProvider>(
         &mut self,
         provider: &Provider,
         signer: &impl Signer,
@@ -54,9 +56,11 @@ impl MlsGroup {
         provider
             .storage()
             .write_group_state(self.group_id(), &self.group_state)
+            .await
             .map_err(SelfUpdateError::StorageError)?;
         self.group
             .store(provider.storage())
+            .await
             .map_err(SelfUpdateError::StorageError)?;
 
         Ok((

openmls/src/treesync/node/encryption_keys.rs

@@ -174,13 +174,14 @@ impl EncryptionKeyPair {
     /// of the key package or the epoch encryption key pairs.
     ///
     /// Returns `None` if the keypair cannot be read from the store.
-    pub(crate) fn read(
+    pub(crate) async fn read(
         provider: &impl OpenMlsProvider,
         encryption_key: &EncryptionKey,
     ) -> Option<EncryptionKeyPair> {
         provider
             .storage()
             .encryption_key_pair(encryption_key)
+            .await
             .ok()
             .flatten()
     }