Merge branch 'main' into admin-frontend/self-host-env

Cargo.lock

@@ -1312,6 +1312,7 @@ dependencies = [
  "collab-folder",
  "database-entity",
  "dotenv",
+ "futures",
  "gotrue",
  "image",
  "lazy_static",

libs/client-api-test-util/Cargo.toml

@@ -31,6 +31,7 @@ dotenv = "0.15.0"
 reqwest = "0.11.23"
 gotrue.workspace = true
 websocket.workspace = true
+futures = "0.3.30"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 web-sys = { version = "0.3", features = ["console"] }

libs/client-api-test-util/src/test_client.rs

@@ -500,7 +500,7 @@ impl TestClient {
     );
 
     collab.lock().add_plugin(Arc::new(sync_plugin));
-    collab.lock().initialize().await;
+    futures::executor::block_on(collab.lock().initialize());
     let test_collab = TestCollab { origin, collab };
     self
       .collab_by_object_id

libs/realtime/src/collaborate/server.rs

@@ -483,6 +483,9 @@ impl CollabClientStream {
         if can_sink {
           // Send the message to websocket client actor
           client_ws_sink.do_send(msg.into());
+        } else {
+          // when then client is not allowed to receive the message
+          tokio::time::sleep(Duration::from_secs(2)).await;
         }
       }
     });
@@ -496,6 +499,9 @@ impl CollabClientStream {
       while let Some(Ok(Ok(RealtimeMessage::Collab(msg)))) = stream_rx.next().await {
         if stream_filter(&cloned_object_id, &msg).await {
           let _ = tx.send(Ok(msg)).await;
+        } else {
+          // when then client is not allowed to receive the message
+          tokio::time::sleep(Duration::from_secs(2)).await;
         }
       }
     });

tests/collab/single_device_edit.rs

@@ -1,9 +1,8 @@
-use collab_entity::CollabType;
-
 use crate::collab::util::{generate_random_string, make_big_collab_doc_state};
+use assert_json_diff::assert_json_eq;
 use client_api_test_util::*;
+use collab_entity::CollabType;
 use database_entity::dto::AFAccessLevel;
-
 use serde_json::json;
 use uuid::Uuid;
 
@@ -406,3 +405,51 @@ async fn multiple_collab_edit_test() {
   )
   .await;
 }
+
+#[tokio::test]
+async fn concurrent_device_edit_test() {
+  let mut tasks = Vec::new();
+  for _i in 0..20 {
+    let task = tokio::spawn(async move {
+      let collab_type = CollabType::Document;
+      let mut test_client = TestClient::new_user().await;
+
+      let workspace_id = test_client.workspace_id().await;
+      let object_id = Uuid::new_v4().to_string();
+
+      test_client
+        .open_collab(&workspace_id, &object_id, collab_type.clone())
+        .await;
+
+      let random_str = generate_random_string(200);
+      test_client
+        .collab_by_object_id
+        .get_mut(&object_id)
+        .unwrap()
+        .collab
+        .lock()
+        .insert("string", random_str.clone());
+      let expected_json = json!({
+        "string": random_str
+      });
+
+      test_client.wait_object_sync_complete(&object_id).await;
+      (
+        expected_json,
+        test_client
+          .collab_by_object_id
+          .get(&object_id)
+          .unwrap()
+          .collab
+          .to_json_value(),
+      )
+    });
+    tasks.push(task);
+  }
+
+  let results = futures::future::join_all(tasks).await;
+  for result in results {
+    let (expected_json, json) = result.unwrap();
+    assert_json_eq!(expected_json, json);
+  }
+}

tests/user/sign_up.rs

@@ -1,6 +1,7 @@
 use app_error::ErrorCode;
 use client_api_test_util::*;
 use gotrue_entity::dto::AuthProvider;
+use std::time::Duration;
 
 #[tokio::test]
 async fn sign_up_success() {
@@ -59,3 +60,20 @@ async fn sign_up_oauth_not_available() {
     ErrorCode::InvalidOAuthProvider
   );
 }
+
+#[tokio::test]
+async fn concurrent_user_sign_up_test() {
+  let mut tasks = Vec::new();
+  for _i in 0..50 {
+    let task = tokio::spawn(async move {
+      let _ = TestClient::new_user().await;
+      tokio::time::sleep(Duration::from_millis(300)).await;
+    });
+    tasks.push(task);
+  }
+
+  let results = futures::future::join_all(tasks).await;
+  for result in results {
+    assert!(result.is_ok(), "Task completed successfully");
+  }
+}