fix: invalid message id (#762)

* chore: add logs

* chore: remove all pending message when server missing update

libs/client-api/src/collab_sync/collab_sink.rs

@@ -172,18 +172,8 @@ where
   }
 
   pub fn clear(&self) {
-    match self.message_queue.try_lock() {
-      None => error!("failed to acquire the lock of the sink"),
-      Some(mut msg_queue) => {
-        msg_queue.clear();
-      },
-    }
-    match self.sending_messages.try_lock() {
-      None => error!("failed to acquire the lock of the flying message"),
-      Some(mut sending_messages) => {
-        sending_messages.clear();
-      },
-    }
+    self.message_queue.lock().clear();
+    self.sending_messages.lock().clear();
   }
 
   pub fn pause(&self) {
@@ -216,6 +206,14 @@ where
 
     // if the message id is not in the sending messages, it means the message is invalid.
     if !sending_messages.contains(&income_message_id) {
+      if cfg!(feature = "sync_verbose_log") {
+        trace!(
+          "{}: sending messages:{:?} not contains {}",
+          self.object.object_id,
+          sending_messages,
+          income_message_id
+        );
+      }
       return Ok(false);
     }
 

libs/client-api/src/collab_sync/collab_stream.rs

@@ -254,6 +254,11 @@ where
       }
 
       if ack_code == AckCode::MissUpdate {
+        // if the ack code is MissUpdate, it means the server has missed some updates. Client need to
+        // use the payload of the current message to calculate missing update. So any existing pending
+        // updates are no long needed.
+        sink.clear();
+
         return Err(SyncError::MissUpdates {
           state_vector_v1: Some(ack.payload.to_vec()),
           reason: MissUpdateReason::ServerMissUpdates,

libs/collab-rt-entity/src/server_message.rs

@@ -245,11 +245,12 @@ impl AckMeta {
 impl Display for CollabAck {
   fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
     f.write_fmt(format_args!(
-      "ack: [uid:{}|oid:{}|msg_id:{:?}|len:{}]",
+      "ack: [uid:{}|oid:{}|msg_id:{:?}|len:{}|code:{}]",
       self.origin.client_user_id().unwrap_or(0),
       self.object_id,
       self.msg_id,
       self.payload.len(),
+      self.code,
     ))
   }
 }