fix: support index document in the background, fix stack overflow whe…

…n calling rayon::spawn (#1099)

* chore: batch index

* chore: format log

* chore: index workspace

* chore: fix stack overflow

* chore: background index

* chore: clippy

* chore: filter tasks

* chore: clippy

* chore: add metrics

* chore: fix test

Cargo.toml

@@ -156,6 +156,7 @@ base64.workspace = true
 md5.workspace = true
 nanoid = "0.4.0"
 http.workspace = true
+indexer.workspace = true
 
 [dev-dependencies]
 once_cell = "1.19.0"
@@ -176,7 +177,6 @@ collab-rt-entity = { path = "libs/collab-rt-entity" }
 hex = "0.4.3"
 unicode-normalization = "0.1.24"
 
-
 [[bin]]
 name = "appflowy_cloud"
 path = "src/main.rs"
@@ -221,9 +221,11 @@ members = [
   "xtask",
   "libs/tonic-proto",
   "libs/mailer",
+  "libs/indexer",
 ]
 
 [workspace.dependencies]
+indexer = { path = "libs/indexer" }
 collab-rt-entity = { path = "libs/collab-rt-entity" }
 collab-rt-protocol = { path = "libs/collab-rt-protocol" }
 database = { path = "libs/database" }

deploy.env

@@ -157,6 +157,7 @@ APPFLOWY_LOCAL_AI_TEST_ENABLED=false
 APPFLOWY_INDEXER_ENABLED=true
 APPFLOWY_INDEXER_DATABASE_URL=postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB}
 APPFLOWY_INDEXER_REDIS_URL=redis://${REDIS_HOST}:${REDIS_PORT}
+APPFLOWY_INDEXER_EMBEDDING_BUFFER_SIZE=5000
 
 # AppFlowy Collaborate
 APPFLOWY_COLLABORATE_MULTI_THREAD=false

dev.env

@@ -124,6 +124,7 @@ APPFLOWY_LOCAL_AI_TEST_ENABLED=false
 APPFLOWY_INDEXER_ENABLED=true
 APPFLOWY_INDEXER_DATABASE_URL=postgres://postgres:password@postgres:5432/postgres
 APPFLOWY_INDEXER_REDIS_URL=redis://redis:6379
+APPFLOWY_INDEXER_EMBEDDING_BUFFER_SIZE=5000
 
 # AppFlowy Collaborate
 APPFLOWY_COLLABORATE_MULTI_THREAD=false

libs/app-error/src/lib.rs

@@ -182,6 +182,9 @@ pub enum AppError {
   #[error("Decode update error: {0}")]
   DecodeUpdateError(String),
 
+  #[error("{0}")]
+  ActionTimeout(String),
+
   #[error("Apply update error:{0}")]
   ApplyUpdateError(String),
 }
@@ -263,6 +266,7 @@ impl AppError {
       AppError::ServiceTemporaryUnavailable(_) => ErrorCode::ServiceTemporaryUnavailable,
       AppError::DecodeUpdateError(_) => ErrorCode::DecodeUpdateError,
       AppError::ApplyUpdateError(_) => ErrorCode::ApplyUpdateError,
+      AppError::ActionTimeout(_) => ErrorCode::ActionTimeout,
     }
   }
 }
@@ -316,6 +320,7 @@ impl From<sqlx::Error> for AppError {
       sqlx::Error::RowNotFound => {
         AppError::RecordNotFound(format!("Record not exist in db. {})", msg))
       },
+      sqlx::Error::PoolTimedOut => AppError::ActionTimeout(value.to_string()),
       _ => AppError::SqlxError(msg),
     }
   }
@@ -424,6 +429,7 @@ pub enum ErrorCode {
   ServiceTemporaryUnavailable = 1054,
   DecodeUpdateError = 1055,
   ApplyUpdateError = 1056,
+  ActionTimeout = 1057,
 }
 
 impl ErrorCode {

libs/collab-rt-protocol/src/data_validation.rs

@@ -6,14 +6,8 @@ use collab::preclude::Collab;
 use collab_entity::CollabType;
 use tracing::instrument;
 
-#[instrument(level = "trace", skip(data), fields(len = %data.len()))]
 #[inline]
-pub async fn spawn_blocking_validate_encode_collab(
-  object_id: &str,
-  data: &[u8],
-  collab_type: &CollabType,
-) -> Result<(), Error> {
-  let collab_type = collab_type.clone();
+pub async fn collab_from_encode_collab(object_id: &str, data: &[u8]) -> Result<Collab, Error> {
   let object_id = object_id.to_string();
   let data = data.to_vec();
 
@@ -27,28 +21,19 @@ pub async fn spawn_blocking_validate_encode_collab(
       false,
     )?;
 
-    collab_type.validate_require_data(&collab)?;
-    Ok::<(), Error>(())
+    Ok::<_, Error>(collab)
   })
   .await?
 }
 
 #[instrument(level = "trace", skip(data), fields(len = %data.len()))]
 #[inline]
-pub fn validate_encode_collab(
+pub async fn validate_encode_collab(
   object_id: &str,
   data: &[u8],
   collab_type: &CollabType,
 ) -> Result<(), Error> {
-  let encoded_collab = EncodedCollab::decode_from_bytes(data)?;
-  let collab = Collab::new_with_source(
-    CollabOrigin::Empty,
-    object_id,
-    DataSource::DocStateV1(encoded_collab.doc_state.to_vec()),
-    vec![],
-    false,
-  )?;
-
+  let collab = collab_from_encode_collab(object_id, data).await?;
   collab_type.validate_require_data(&collab)?;
   Ok::<(), Error>(())
 }