chore: decode encode collab using spawn blocking (#781)

services/appflowy-collaborate/src/collab/decode_util.rs

@@ -0,0 +1,16 @@
+use anyhow::anyhow;
+use app_error::AppError;
+use collab::entity::EncodedCollab;
+
+#[inline]
+pub(crate) async fn encode_collab_from_bytes(bytes: Vec<u8>) -> Result<EncodedCollab, AppError> {
+  // FIXME: Implement metrics to determine the appropriate data size for decoding in a blocking task.
+  tokio::task::spawn_blocking(move || match EncodedCollab::decode_from_bytes(&bytes) {
+    Ok(encoded_collab) => Ok(encoded_collab),
+    Err(err) => Err(AppError::Internal(anyhow!(
+      "Failed to decode collab from bytes: {:?}",
+      err
+    ))),
+  })
+  .await?
+}

services/appflowy-collaborate/src/collab/disk_cache.rs

@@ -1,7 +1,6 @@
 use std::collections::HashMap;
 use std::time::Duration;
 
-use anyhow::anyhow;
 use collab::entity::EncodedCollab;
 use collab_entity::CollabType;
 use sqlx::{Error, PgPool, Transaction};
@@ -9,6 +8,7 @@ use tokio::time::sleep;
 use tracing::{event, instrument, Level};
 use uuid::Uuid;
 
+use crate::collab::decode_util::encode_collab_from_bytes;
 use app_error::AppError;
 use database::collab::{
   batch_select_collab_blob, insert_into_af_collab, is_collab_exists, select_blob_from_af_collab,
@@ -92,12 +92,7 @@ impl CollabDiskCache {
 
       match result {
         Ok(data) => {
-          return tokio::task::spawn_blocking(move || {
+          return encode_collab_from_bytes(data).await;
-            EncodedCollab::decode_from_bytes(&data).map_err(|err| {
-              AppError::Internal(anyhow!("fail to decode data to EncodedCollab: {:?}", err))
-            })
-          })
-          .await?;
         },
         Err(e) => {
           match e {

services/appflowy-collaborate/src/collab/mem_cache.rs

@@ -3,11 +3,11 @@ use collab::entity::EncodedCollab;
 use redis::{pipe, AsyncCommands};
 use tracing::{error, instrument, trace};
 
+use crate::collab::decode_util::encode_collab_from_bytes;
+use crate::state::RedisConnectionManager;
 use app_error::AppError;
 use database::collab::CollabMetadata;
 
-use crate::state::RedisConnectionManager;
-
 const SEVEN_DAYS: i64 = 604800;
 const ONE_MONTH: u64 = 2592000;
 #[derive(Clone)]
@@ -96,13 +96,7 @@ impl CollabMemCache {
   #[instrument(level = "trace", skip_all)]
   pub async fn get_encode_collab(&self, object_id: &str) -> Option<EncodedCollab> {
     match self.get_encode_collab_data(object_id).await {
-      Some(bytes) => match EncodedCollab::decode_from_bytes(&bytes) {
+      Some(bytes) => encode_collab_from_bytes(bytes).await.ok(),
-        Ok(encoded_collab) => Some(encoded_collab),
-        Err(err) => {
-          error!("Failed to decode collab from redis cache bytes: {:?}", err);
-          None
-        },
-      },
       None => {
         trace!(
           "No encoded collab found in cache for object_id: {}",

services/appflowy-collaborate/src/collab/mod.rs

@@ -1,5 +1,6 @@
 pub mod access_control;
 pub mod cache;
+mod decode_util;
 pub mod disk_cache;
 pub mod mem_cache;
 pub mod notification;
@@ -7,4 +8,5 @@ pub mod queue;
 mod queue_redis_ops;
 pub mod storage;
 pub mod validator;
+
 pub use queue_redis_ops::{PendingWrite, RedisSortedSet, WritePriority};

src/api/workspace.rs

@@ -920,22 +920,27 @@ async fn update_collab_handler(
       .can_index_workspace(&workspace_id)
       .await?
     {
-      let encoded = EncodedCollab::decode_from_bytes(&params.encoded_collab_v1).map_err(|err| {
+      let (encoded, mut mut_params) = tokio::task::spawn_blocking(move || {
-        AppError::InvalidRequest(format!(
+        EncodedCollab::decode_from_bytes(&params.encoded_collab_v1)
-          "Failed to decode collab `{}`: {}",
+          .map(|encoded_collab| (encoded_collab, params))
-          params.object_id, err
+          .map_err(|err| AppError::InvalidRequest(format!("Failed to decode collab `{}", err)))
-        ))
+      })
-      })?;
+      .await
-      match indexer.index(&params.object_id, encoded).await {
+      .map_err(|err| AppError::Internal(err.into()))??;
-        Ok(embeddings) => params.embeddings = embeddings,
+
+      match indexer.index(&mut_params.object_id, encoded).await {
+        Ok(embeddings) => mut_params.embeddings = embeddings,
         Err(err) => tracing::warn!(
           "failed to fetch embeddings for document {}: {}",
-          params.object_id,
+          mut_params.object_id,
           err
         ),
       }
+
+      params = mut_params;
     }
   }
+
   state
     .collab_access_control_storage
     .insert_or_update_collab(&workspace_id, &uid, params, false)