chore: calculate the compact length after receiving ack

Author: appflowy

frontend/rust-lib/Cargo.lock

@@ -29,7 +29,9 @@ impl AppFlowyDocumentEditor {
         mut rev_manager: RevisionManager<Arc<ConnectionPool>>,
         cloud_service: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
-        let document = rev_manager.load::<DocumentRevisionSerde>(Some(cloud_service)).await?;
+        let document = rev_manager
+            .initialize::<DocumentRevisionSerde>(Some(cloud_service))
+            .await?;
         let rev_manager = Arc::new(rev_manager);
         let command_sender = spawn_edit_queue(user, rev_manager.clone(), document);
         let doc_id = doc_id.to_string();

frontend/rust-lib/flowy-document/src/editor/editor.rs

@@ -45,7 +45,7 @@ impl DeltaDocumentEditor {
         cloud_service: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
         let document = rev_manager
-            .load::<DeltaDocumentRevisionSerde>(Some(cloud_service))
+            .initialize::<DeltaDocumentRevisionSerde>(Some(cloud_service))
             .await?;
         let operations = DeltaTextOperations::from_bytes(&document.content)?;
         let rev_manager = Arc::new(rev_manager);

frontend/rust-lib/flowy-document/src/old_editor/editor.rs

@@ -38,7 +38,9 @@ impl FolderEditor {
         let cloud = Arc::new(FolderRevisionCloudService {
             token: token.to_string(),
         });
-        let folder = Arc::new(RwLock::new(rev_manager.load::<FolderRevisionSerde>(Some(cloud)).await?));
+        let folder = Arc::new(RwLock::new(
+            rev_manager.initialize::<FolderRevisionSerde>(Some(cloud)).await?,
+        ));
         let rev_manager = Arc::new(rev_manager);
 
         #[cfg(feature = "sync")]

frontend/rust-lib/flowy-folder/src/services/folder_editor.rs

@@ -34,7 +34,7 @@ impl GridBlockRevisionEditor {
         let cloud = Arc::new(GridBlockRevisionCloudService {
             token: token.to_owned(),
         });
-        let block_revision_pad = rev_manager.load::<GridBlockRevisionSerde>(Some(cloud)).await?;
+        let block_revision_pad = rev_manager.initialize::<GridBlockRevisionSerde>(Some(cloud)).await?;
         let pad = Arc::new(RwLock::new(block_revision_pad));
         let rev_manager = Arc::new(rev_manager);
         let user_id = user_id.to_owned();

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

@@ -60,7 +60,7 @@ impl GridRevisionEditor {
     ) -> FlowyResult<Arc<Self>> {
         let token = user.token()?;
         let cloud = Arc::new(GridRevisionCloudService { token });
-        let grid_pad = rev_manager.load::<GridRevisionSerde>(Some(cloud)).await?;
+        let grid_pad = rev_manager.initialize::<GridRevisionSerde>(Some(cloud)).await?;
         let rev_manager = Arc::new(rev_manager);
         let grid_pad = Arc::new(RwLock::new(grid_pad));
 

frontend/rust-lib/flowy-grid/src/services/grid_editor.rs

@@ -55,7 +55,7 @@ impl GridViewRevisionEditor {
         let cloud = Arc::new(GridViewRevisionCloudService {
             token: token.to_owned(),
         });
-        let view_revision_pad = rev_manager.load::<GridViewRevisionSerde>(Some(cloud)).await?;
+        let view_revision_pad = rev_manager.initialize::<GridViewRevisionSerde>(Some(cloud)).await?;
         let pad = Arc::new(RwLock::new(view_revision_pad));
         let rev_manager = Arc::new(rev_manager);
         let group_controller = new_group_controller(

frontend/rust-lib/flowy-grid/src/services/grid_view_editor.rs

@@ -23,6 +23,7 @@ serde_json = {version = "1.0"}
 
 [dev-dependencies]
 nanoid = "0.4.0"
+flowy-revision = {path = ".", features = ["flowy_unit_test"]}
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0" }
 parking_lot = "0.11"

frontend/rust-lib/flowy-revision/Cargo.toml

@@ -108,7 +108,7 @@ impl<Connection: 'static> RevisionManager<Connection> {
     }
 
     #[tracing::instrument(level = "debug", skip_all, fields(object_id) err)]
-    pub async fn load<B>(&mut self, cloud: Option<Arc<dyn RevisionCloudService>>) -> FlowyResult<B::Output>
+    pub async fn initialize<B>(&mut self, cloud: Option<Arc<dyn RevisionCloudService>>) -> FlowyResult<B::Output>
     where
         B: RevisionObjectDeserializer,
     {
@@ -199,6 +199,10 @@ impl<Connection: 'static> RevisionManager<Connection> {
         self.rev_persistence.number_of_sync_records()
     }
 
+    pub fn number_of_revisions_in_disk(&self) -> usize {
+        self.rev_persistence.number_of_records_in_disk()
+    }
+
     pub async fn get_revisions_in_range(&self, range: RevisionRange) -> Result<Vec<Revision>, FlowyError> {
         let revisions = self.rev_persistence.revisions_in_range(&range).await?;
         Ok(revisions)
@@ -230,13 +234,16 @@ impl<Connection: 'static> WSDataProviderDataSource for Arc<RevisionManager<Conne
 }
 
 #[cfg(feature = "flowy_unit_test")]
-impl<Connection> RevisionManager<Connection> {
+impl<Connection: 'static> RevisionManager<Connection> {
     pub async fn revision_cache(&self) -> Arc<RevisionPersistence<Connection>> {
         self.rev_persistence.clone()
     }
     pub fn ack_notify(&self) -> tokio::sync::broadcast::Receiver<i64> {
         self.rev_ack_notifier.subscribe()
     }
+    pub fn get_all_revision_records(&self) -> FlowyResult<Vec<crate::disk::SyncRecord>> {
+        self.rev_persistence.load_all_records(&self.object_id)
+    }
 }
 
 pub struct RevisionLoader<Connection> {
@@ -248,7 +255,7 @@ pub struct RevisionLoader<Connection> {
 
 impl<Connection: 'static> RevisionLoader<Connection> {
     pub async fn load(&self) -> Result<(Vec<Revision>, i64), FlowyError> {
-        let records = self.rev_persistence.batch_get(&self.object_id)?;
+        let records = self.rev_persistence.load_all_records(&self.object_id)?;
         let revisions: Vec<Revision>;
         let mut rev_id = 0;
         if records.is_empty() && self.cloud.is_some() {
@@ -282,7 +289,7 @@ impl<Connection: 'static> RevisionLoader<Connection> {
     }
 
     pub async fn load_revisions(&self) -> Result<Vec<Revision>, FlowyError> {
-        let records = self.rev_persistence.batch_get(&self.object_id)?;
+        let records = self.rev_persistence.load_all_records(&self.object_id)?;
         let revisions = records.into_iter().map(|record| record.revision).collect::<_>();
         Ok(revisions)
     }

frontend/rust-lib/flowy-revision/src/rev_manager.rs

@@ -14,6 +14,7 @@ use tokio::task::spawn_blocking;
 
 pub const REVISION_WRITE_INTERVAL_IN_MILLIS: u64 = 600;
 
+#[derive(Clone)]
 pub struct RevisionPersistenceConfiguration {
     merge_threshold: usize,
 }
@@ -24,14 +25,14 @@ impl RevisionPersistenceConfiguration {
         if merge_threshold > 1 {
             Self { merge_threshold }
         } else {
-            Self { merge_threshold: 2 }
+            Self { merge_threshold: 100 }
         }
     }
 }
 
 impl std::default::Default for RevisionPersistenceConfiguration {
     fn default() -> Self {
-        Self { merge_threshold: 2 }
+        Self { merge_threshold: 100 }
     }
 }
 
@@ -93,7 +94,7 @@ where
     pub(crate) async fn sync_revision(&self, revision: &Revision) -> FlowyResult<()> {
         tracing::Span::current().record("rev_id", &revision.rev_id);
         self.add(revision.clone(), RevisionState::Sync, false).await?;
-        self.sync_seq.write().await.dry_push(revision.rev_id)?;
+        self.sync_seq.write().await.recv(revision.rev_id)?;
         Ok(())
     }
 
@@ -105,13 +106,17 @@ where
         rev_compress: &Arc<dyn RevisionMergeable + 'a>,
     ) -> FlowyResult<i64> {
         let mut sync_seq = self.sync_seq.write().await;
-        let step = sync_seq.step;
+        let compact_length = sync_seq.compact_length;
 
-        // Before the new_revision pushed into the sync_seq, we check if the current `step` of the
-        // sync_seq is less equal or greater than the merge threshold. If yes, it's need to merged
+        // Before the new_revision is pushed into the sync_seq, we check if the current `step` of the
+        // sync_seq is less equal to or greater than the merge threshold. If yes, it's needs to merged
         // with the new_revision into one revision.
-        if step >= self.configuration.merge_threshold - 1 {
-            let compact_seq = sync_seq.compact();
+        let mut compact_seq = VecDeque::default();
+        // tracing::info!("{}", compact_seq)
+        if compact_length >= self.configuration.merge_threshold - 1 {
+            compact_seq.extend(sync_seq.compact());
+        }
+        if !compact_seq.is_empty() {
             let range = RevisionRange {
                 start: *compact_seq.front().unwrap(),
                 end: *compact_seq.back().unwrap(),
@@ -127,15 +132,15 @@ where
             let merged_revision = rev_compress.merge_revisions(&self.user_id, &self.object_id, revisions)?;
             let rev_id = merged_revision.rev_id;
             tracing::Span::current().record("rev_id", &merged_revision.rev_id);
-            let _ = sync_seq.dry_push(merged_revision.rev_id)?;
+            let _ = sync_seq.recv(merged_revision.rev_id)?;
 
             // replace the revisions in range with compact revision
             self.compact(&range, merged_revision).await?;
             Ok(rev_id)
         } else {
             tracing::Span::current().record("rev_id", &new_revision.rev_id);
             self.add(new_revision.clone(), RevisionState::Sync, true).await?;
-            sync_seq.push(new_revision.rev_id)?;
+            sync_seq.merge_recv(new_revision.rev_id)?;
             Ok(new_revision.rev_id)
         }
     }
@@ -163,6 +168,16 @@ where
         self.memory_cache.number_of_sync_records()
     }
 
+    pub(crate) fn number_of_records_in_disk(&self) -> usize {
+        match self.disk_cache.read_revision_records(&self.object_id, None) {
+            Ok(records) => records.len(),
+            Err(e) => {
+                tracing::error!("Read revision records failed: {:?}", e);
+                0
+            }
+        }
+    }
+
     /// The cache gets reset while it conflicts with the remote revisions.
     #[tracing::instrument(level = "trace", skip(self, revisions), err)]
     pub(crate) async fn reset(&self, revisions: Vec<Revision>) -> FlowyResult<()> {
@@ -228,8 +243,8 @@ where
         }
     }
 
-    pub fn batch_get(&self, doc_id: &str) -> FlowyResult<Vec<SyncRecord>> {
-        self.disk_cache.read_revision_records(doc_id, None)
+    pub fn load_all_records(&self, object_id: &str) -> FlowyResult<Vec<SyncRecord>> {
+        self.disk_cache.read_revision_records(object_id, None)
     }
 
     // Read the revision which rev_id >= range.start && rev_id <= range.end
@@ -289,26 +304,31 @@ impl<C> RevisionMemoryCacheDelegate for Arc<dyn RevisionDiskCache<C, Error = Flo
 #[derive(Default)]
 struct DeferSyncSequence {
     rev_ids: VecDeque<i64>,
-    start: Option<usize>,
-    step: usize,
+    compact_index: Option<usize>,
+    compact_length: usize,
 }
 
 impl DeferSyncSequence {
     fn new() -> Self {
         DeferSyncSequence::default()
     }
 
-    fn push(&mut self, new_rev_id: i64) -> FlowyResult<()> {
-        let _ = self.dry_push(new_rev_id)?;
+    /// Pushes the new_rev_id to the end of the list and marks this new_rev_id is mergeable.
+    ///
+    /// When calling `compact` method, it will return a list of revision ids started from
+    /// the `compact_start_pos`, and ends with the `compact_length`.
+    fn merge_recv(&mut self, new_rev_id: i64) -> FlowyResult<()> {
+        let _ = self.recv(new_rev_id)?;
 
-        self.step += 1;
-        if self.start.is_none() && !self.rev_ids.is_empty() {
-            self.start = Some(self.rev_ids.len() - 1);
+        self.compact_length += 1;
+        if self.compact_index.is_none() && !self.rev_ids.is_empty() {
+            self.compact_index = Some(self.rev_ids.len() - 1);
         }
         Ok(())
     }
 
-    fn dry_push(&mut self, new_rev_id: i64) -> FlowyResult<()> {
+    /// Pushes the new_rev_id to the end of the list.
+    fn recv(&mut self, new_rev_id: i64) -> FlowyResult<()> {
         // The last revision's rev_id must be greater than the new one.
         if let Some(rev_id) = self.rev_ids.back() {
             if *rev_id >= new_rev_id {
@@ -321,6 +341,7 @@ impl DeferSyncSequence {
         Ok(())
     }
 
+    /// Removes the rev_id from the list
     fn ack(&mut self, rev_id: &i64) -> FlowyResult<()> {
         let cur_rev_id = self.rev_ids.front().cloned();
         if let Some(pop_rev_id) = cur_rev_id {
@@ -331,7 +352,20 @@ impl DeferSyncSequence {
                 );
                 return Err(FlowyError::internal().context(desc));
             }
-            let _ = self.rev_ids.pop_front();
+
+            let mut compact_rev_id = None;
+            if let Some(compact_index) = self.compact_index {
+                compact_rev_id = self.rev_ids.get(compact_index).cloned();
+            }
+
+            let pop_rev_id = self.rev_ids.pop_front();
+            if let (Some(compact_rev_id), Some(pop_rev_id)) = (compact_rev_id, pop_rev_id) {
+                if compact_rev_id <= pop_rev_id {
+                    if self.compact_length > 0 {
+                        self.compact_length -= 1;
+                    }
+                }
+            }
         }
         Ok(())
     }
@@ -341,28 +375,22 @@ impl DeferSyncSequence {
     }
 
     fn clear(&mut self) {
-        self.start = None;
-        self.step = 0;
+        self.compact_index = None;
+        self.compact_length = 0;
         self.rev_ids.clear();
     }
 
     // Compact the rev_ids into one except the current synchronizing rev_id.
     fn compact(&mut self) -> VecDeque<i64> {
-        if self.start.is_none() {
-            return VecDeque::default();
+        let mut compact_seq = VecDeque::with_capacity(self.rev_ids.len());
+        if let Some(start) = self.compact_index {
+            if start < self.rev_ids.len() {
+                let seq = self.rev_ids.split_off(start);
+                compact_seq.extend(seq);
+            }
         }
-
-        let start = self.start.unwrap();
-        let compact_seq = self.rev_ids.split_off(start);
-        self.start = None;
-        self.step = 0;
+        self.compact_index = None;
+        self.compact_length = 0;
         compact_seq
-
-        // let mut new_seq = self.rev_ids.clone();
-        // let mut drained = new_seq.drain(1..).collect::<VecDeque<_>>();
-        //
-        // let start = drained.pop_front()?;
-        // let end = drained.pop_back().unwrap_or(start);
-        // Some((RevisionRange { start, end }, new_seq))
     }
 }