fix: potential fail to read the data of the row (#1793)

frontend/rust-lib/flowy-client-sync/src/client_database/block_revision_pad.rs

@@ -27,7 +27,7 @@ impl std::ops::Deref for GridBlockRevisionPad {
 }
 
 impl GridBlockRevisionPad {
-    pub async fn duplicate_data(&self, duplicated_block_id: &str) -> DatabaseBlockRevision {
+    pub fn duplicate_data(&self, duplicated_block_id: &str) -> DatabaseBlockRevision {
         let duplicated_rows = self
             .block
             .rows

frontend/rust-lib/flowy-database/src/event_handler.rs

@@ -375,6 +375,7 @@ pub(crate) async fn update_select_option_handler(
     let changeset: SelectOptionChangeset = data.into_inner().try_into()?;
     let editor = manager.get_database_editor(&changeset.cell_path.database_id).await?;
     let field_id = changeset.cell_path.field_id.clone();
+    let (tx, rx) = tokio::sync::oneshot::channel();
     editor
         .modify_field_rev(&field_id, |field_rev| {
             let mut type_option = select_type_option_from_field_rev(field_rev)?;
@@ -403,27 +404,24 @@ pub(crate) async fn update_select_option_handler(
             if is_changed.is_some() {
                 field_rev.insert_type_option(&*type_option);
             }
-
-            if let Some(cell_changeset_str) = cell_changeset_str {
-                let cloned_editor = editor.clone();
-                tokio::spawn(async move {
-                    match cloned_editor
-                        .update_cell_with_changeset(
-                            &changeset.cell_path.row_id,
-                            &changeset.cell_path.field_id,
-                            cell_changeset_str,
-                        )
-                        .await
-                    {
-                        Ok(_) => {}
-                        Err(e) => tracing::error!("{}", e),
-                    }
-                });
-            }
+            let _ = tx.send(cell_changeset_str);
             Ok(is_changed)
         })
         .await?;
 
+    if let Ok(Some(cell_changeset_str)) = rx.await {
+        match editor
+            .update_cell_with_changeset(
+                &changeset.cell_path.row_id,
+                &changeset.cell_path.field_id,
+                cell_changeset_str,
+            )
+            .await
+        {
+            Ok(_) => {}
+            Err(e) => tracing::error!("{}", e),
+        }
+    }
     Ok(())
 }
 

frontend/rust-lib/flowy-database/src/services/block_editor.rs

@@ -1,3 +1,4 @@
+use crate::services::retry::GetRowDataRetryAction;
 use bytes::Bytes;
 use flowy_client_sync::client_database::{GridBlockRevisionChangeset, GridBlockRevisionPad};
 use flowy_client_sync::make_operations_from_revisions;
@@ -8,11 +9,14 @@ use flowy_revision::{
 use flowy_sqlite::ConnectionPool;
 use grid_model::{CellRevision, DatabaseBlockRevision, RowChangeset, RowRevision};
 use lib_infra::future::FutureResult;
+use lib_infra::retry::spawn_retry;
 use lib_ot::core::EmptyAttributes;
+use parking_lot::RwLock;
 use revision_model::Revision;
 use std::borrow::Cow;
 use std::sync::Arc;
-use tokio::sync::RwLock;
+use std::time::Duration;
+// use tokio::sync::RwLock;
 
 pub struct DatabaseBlockRevisionEditor {
     #[allow(dead_code)]
@@ -53,7 +57,7 @@ impl DatabaseBlockRevisionEditor {
     }
 
     pub async fn duplicate_block(&self, duplicated_block_id: &str) -> DatabaseBlockRevision {
-        self.pad.read().await.duplicate_data(duplicated_block_id).await
+        self.pad.read().duplicate_data(duplicated_block_id)
     }
 
     /// Create a row after the the with prev_row_id. If prev_row_id is None, the row will be appended to the list
@@ -108,20 +112,30 @@ impl DatabaseBlockRevisionEditor {
     }
 
     pub async fn index_of_row(&self, row_id: &str) -> Option<usize> {
-        self.pad.read().await.index_of_row(row_id)
+        self.pad.read().index_of_row(row_id)
     }
 
     pub async fn number_of_rows(&self) -> i32 {
-        self.pad.read().await.rows.len() as i32
+        self.pad.read().rows.len() as i32
     }
 
     pub async fn get_row_rev(&self, row_id: &str) -> FlowyResult<Option<(usize, Arc<RowRevision>)>> {
-        if self.pad.try_read().is_err() {
-            tracing::error!("Required grid block read lock failed");
-            Ok(None)
+        let duration = Duration::from_millis(300);
+        if let Some(pad) = self.pad.try_read_for(duration) {
+            Ok(pad.get_row_rev(row_id))
         } else {
-            let row_rev = self.pad.read().await.get_row_rev(row_id);
-            Ok(row_rev)
+            tracing::error!("Required grid block read lock failed, retrying");
+            let retry = GetRowDataRetryAction {
+                row_id: row_id.to_owned(),
+                pad: self.pad.clone(),
+            };
+            match spawn_retry(3, 300, retry).await {
+                Ok(value) => Ok(value),
+                Err(err) => {
+                    tracing::error!("Read row revision failed with: {}", err);
+                    Ok(None)
+                }
+            }
         }
     }
 
@@ -129,7 +143,7 @@ impl DatabaseBlockRevisionEditor {
     where
         T: AsRef<str> + ToOwned + ?Sized,
     {
-        let row_revs = self.pad.read().await.get_row_revs(row_ids)?;
+        let row_revs = self.pad.read().get_row_revs(row_ids)?;
         Ok(row_revs)
     }
 
@@ -138,7 +152,7 @@ impl DatabaseBlockRevisionEditor {
         field_id: &str,
         row_ids: Option<Vec<Cow<'_, String>>>,
     ) -> FlowyResult<Vec<CellRevision>> {
-        let cell_revs = self.pad.read().await.get_cell_revs(field_id, row_ids)?;
+        let cell_revs = self.pad.read().get_cell_revs(field_id, row_ids)?;
         Ok(cell_revs)
     }
 
@@ -146,11 +160,11 @@ impl DatabaseBlockRevisionEditor {
     where
         F: for<'a> FnOnce(&'a mut GridBlockRevisionPad) -> FlowyResult<Option<GridBlockRevisionChangeset>>,
     {
-        let mut write_guard = self.pad.write().await;
-        match f(&mut write_guard)? {
+        let changeset = f(&mut self.pad.write())?;
+        match changeset {
             None => {}
-            Some(change) => {
-                self.apply_change(change).await?;
+            Some(changeset) => {
+                self.apply_change(changeset).await?;
             }
         }
         Ok(())

frontend/rust-lib/flowy-database/src/services/mod.rs

@@ -9,6 +9,7 @@ pub mod grid_editor;
 mod grid_editor_trait_impl;
 pub mod group;
 pub mod persistence;
+mod retry;
 pub mod row;
 pub mod setting;
 pub mod sort;

frontend/rust-lib/flowy-database/src/services/retry.rs

@@ -0,0 +1,31 @@
+use flowy_client_sync::client_database::GridBlockRevisionPad;
+use flowy_error::FlowyError;
+use grid_model::RowRevision;
+use lib_infra::retry::Action;
+
+use parking_lot::RwLock;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+
+pub struct GetRowDataRetryAction {
+    pub row_id: String,
+    pub pad: Arc<RwLock<GridBlockRevisionPad>>,
+}
+
+impl Action for GetRowDataRetryAction {
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Item, Self::Error>> + Send + Sync>>;
+    type Item = Option<(usize, Arc<RowRevision>)>;
+    type Error = FlowyError;
+
+    fn run(&mut self) -> Self::Future {
+        let pad = self.pad.clone();
+        let row_id = self.row_id.clone();
+        Box::pin(async move {
+            match pad.try_read() {
+                None => Ok(None),
+                Some(read_guard) => Ok(read_guard.get_row_rev(&row_id)),
+            }
+        })
+    }
+}