chore: detect and re-request missing updates on the server side

Cargo.toml

@@ -285,7 +285,7 @@ pin-project = "1.1.5"
 arc-swap = { version = "1.7" }
 
 # collaboration
-yrs = { version = "0.21.2", features = ["sync"] }
+yrs = { version = "0.21.3", features = ["sync"] }
 collab = { version = "0.2.0" }
 collab-entity = { version = "0.2.0" }
 collab-folder = { version = "0.2.0" }
@@ -306,13 +306,13 @@ debug = true
 [patch.crates-io]
 # It's diffcult to resovle different version with the same crate used in AppFlowy Frontend and the Client-API crate.
 # So using patch to workaround this issue.
-collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "4bd78daec6d4bce5cb0c0be8d2536973aff2861d" }
-collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "4bd78daec6d4bce5cb0c0be8d2536973aff2861d" }
-collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "4bd78daec6d4bce5cb0c0be8d2536973aff2861d" }
-collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "4bd78daec6d4bce5cb0c0be8d2536973aff2861d" }
-collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "4bd78daec6d4bce5cb0c0be8d2536973aff2861d" }
-collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "4bd78daec6d4bce5cb0c0be8d2536973aff2861d" }
-collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "4bd78daec6d4bce5cb0c0be8d2536973aff2861d" }
+collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "477c4912b6e50f4112794773c55363ea51713435" }
+collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "477c4912b6e50f4112794773c55363ea51713435" }
+collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "477c4912b6e50f4112794773c55363ea51713435" }
+collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "477c4912b6e50f4112794773c55363ea51713435" }
+collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "477c4912b6e50f4112794773c55363ea51713435" }
+collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "477c4912b6e50f4112794773c55363ea51713435" }
+collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "477c4912b6e50f4112794773c55363ea51713435" }
 
 [features]
 history = []

services/appflowy-collaborate/src/group/group_init.rs

@@ -673,20 +673,34 @@ impl CollabGroup {
       .await
       .map_err(|err| RTProtocolError::Internal(err.into()))??
     };
-    let state_vector = decoded_update.state_vector();
-    state
-      .persister
-      .send_update(origin.clone(), update, state_vector)
-      .await
-      .map_err(|err| RTProtocolError::Internal(err.into()))?;
-    let elapsed = start.elapsed();
+    let missing_updates = {
+      let state_vector = state.state_vector.read().await;
+      match state_vector.partial_cmp(&decoded_update.state_vector_lower()) {
+        None | Some(std::cmp::Ordering::Less) => Some(state_vector.clone()),
+        _ => None,
+      }
+    };
 
-    state
-      .metrics
-      .apply_update_time
-      .observe(elapsed.as_millis() as f64);
+    if let Some(missing_updates) = missing_updates {
+      let msg = Message::Sync(SyncMessage::SyncStep1(missing_updates));
+      tracing::debug!("subscriber {} send update with missing data", origin);
+      Ok(Some(msg.encode_v1()))
+    } else {
+      let upper_state_vector = decoded_update.state_vector();
+      state
+        .persister
+        .send_update(origin.clone(), update, upper_state_vector)
+        .await
+        .map_err(|err| RTProtocolError::Internal(err.into()))?;
+      let elapsed = start.elapsed();
 
-    Ok(None)
+      state
+        .metrics
+        .apply_update_time
+        .observe(elapsed.as_millis() as f64);
+
+      Ok(None)
+    }
   }
 
   async fn handle_update(