chk
/
AppFlowy-Cloud
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

This commit is contained in:
Bartosz Sypytkowski 2024-10-25 13:11:19 +02:00
parent 923fe7ffde
commit 395424bfd2
1 changed files with 26 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
services/appflowy-collaborate/src/group/group_init.rs
View File

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