AppFlowy/frontend/rust-lib/flowy-sync/src/rev_manager.rs

355 lines
12 KiB
Rust
Raw Normal View History

2022-01-25 11:45:41 +00:00
use crate::RevisionCache;
2021-12-15 15:01:50 +00:00
use flowy_collaboration::{
2022-01-14 07:23:21 +00:00
entities::revision::{RepeatedRevision, Revision, RevisionRange, RevisionState},
util::{pair_rev_id_from_revisions, RevIdCounter},
2021-12-15 15:01:50 +00:00
};
2022-01-14 07:23:21 +00:00
use flowy_error::{FlowyError, FlowyResult};
2021-12-13 05:55:44 +00:00
use lib_infra::future::FutureResult;
2022-01-26 15:29:18 +00:00
2022-01-01 08:16:06 +00:00
use std::{collections::VecDeque, sync::Arc};
2022-01-24 09:35:58 +00:00
use tokio::sync::RwLock;
2022-01-14 07:23:21 +00:00
pub trait RevisionCloudService: Send + Sync {
fn fetch_object(&self, user_id: &str, object_id: &str) -> FutureResult<Vec<Revision>, FlowyError>;
}
2022-01-14 07:23:21 +00:00
pub trait RevisionObjectBuilder: Send + Sync {
type Output;
fn build_with_revisions(object_id: &str, revisions: Vec<Revision>) -> FlowyResult<Self::Output>;
}
2022-01-25 12:37:48 +00:00
pub trait RevisionCompact: Send + Sync {
fn compact_revisions(user_id: &str, object_id: &str, revisions: Vec<Revision>) -> FlowyResult<Revision>;
}
2022-01-14 07:23:21 +00:00
pub struct RevisionManager {
pub object_id: String,
2021-12-09 14:28:11 +00:00
user_id: String,
rev_id_counter: RevIdCounter,
2022-01-25 12:37:48 +00:00
cache: Arc<RwLock<RevisionCacheCompact>>,
2022-01-22 10:48:43 +00:00
#[cfg(feature = "flowy_unit_test")]
2022-01-24 09:35:58 +00:00
revision_ack_notifier: tokio::sync::broadcast::Sender<i64>,
}
2022-01-14 07:23:21 +00:00
impl RevisionManager {
pub fn new(user_id: &str, object_id: &str, revision_cache: Arc<RevisionCache>) -> Self {
2021-10-07 12:46:29 +00:00
let rev_id_counter = RevIdCounter::new(0);
2022-01-26 15:29:18 +00:00
let cache = Arc::new(RwLock::new(RevisionCacheCompact::new(
object_id,
user_id,
revision_cache,
)));
2022-01-22 10:48:43 +00:00
#[cfg(feature = "flowy_unit_test")]
2022-01-24 09:35:58 +00:00
let (revision_ack_notifier, _) = tokio::sync::broadcast::channel(1);
2022-01-22 10:48:43 +00:00
Self {
2022-01-14 07:23:21 +00:00
object_id: object_id.to_string(),
2021-12-09 14:28:11 +00:00
user_id: user_id.to_owned(),
rev_id_counter,
2022-01-25 12:37:48 +00:00
cache,
2022-01-22 10:48:43 +00:00
#[cfg(feature = "flowy_unit_test")]
revision_ack_notifier,
}
}
2022-01-25 12:37:48 +00:00
pub async fn load<B, C>(&mut self, cloud: Arc<dyn RevisionCloudService>) -> FlowyResult<B::Output>
2022-01-14 07:23:21 +00:00
where
2022-01-25 12:37:48 +00:00
B: RevisionObjectBuilder,
C: RevisionCompact,
2022-01-14 07:23:21 +00:00
{
2022-01-23 14:33:47 +00:00
let (revisions, rev_id) = RevisionLoader {
2022-01-14 07:23:21 +00:00
object_id: self.object_id.clone(),
2021-12-17 16:23:26 +00:00
user_id: self.user_id.clone(),
2022-01-14 07:23:21 +00:00
cloud,
2022-01-25 12:37:48 +00:00
cache: self.cache.clone(),
2021-12-17 16:23:26 +00:00
}
2022-01-25 12:37:48 +00:00
.load::<C>()
2021-12-17 16:23:26 +00:00
.await?;
2022-01-23 14:33:47 +00:00
self.rev_id_counter.set(rev_id);
2022-01-25 12:37:48 +00:00
B::build_with_revisions(&self.object_id, revisions)
2021-10-07 12:46:29 +00:00
}
2022-01-01 06:23:58 +00:00
#[tracing::instrument(level = "debug", skip(self, revisions), err)]
2022-01-14 07:23:21 +00:00
pub async fn reset_object(&self, revisions: RepeatedRevision) -> FlowyResult<()> {
2022-01-02 02:34:42 +00:00
let rev_id = pair_rev_id_from_revisions(&revisions).1;
2022-01-26 15:29:18 +00:00
let write_guard = self.cache.write().await;
let _ = write_guard.reset(revisions.into_inner()).await?;
2022-01-02 02:34:42 +00:00
self.rev_id_counter.set(rev_id);
Ok(())
2022-01-01 06:23:58 +00:00
}
2022-01-02 02:34:42 +00:00
#[tracing::instrument(level = "debug", skip(self, revision), err)]
2021-12-14 10:04:51 +00:00
pub async fn add_remote_revision(&self, revision: &Revision) -> Result<(), FlowyError> {
2022-01-01 15:09:13 +00:00
if revision.delta_data.is_empty() {
return Err(FlowyError::internal().context("Delta data should be empty"));
}
2022-01-26 15:29:18 +00:00
let write_guard = self.cache.write().await;
let _ = write_guard.add_ack_revision(revision).await?;
2022-01-07 09:37:11 +00:00
self.rev_id_counter.set(revision.rev_id);
2021-12-13 14:46:35 +00:00
Ok(())
}
2022-01-01 08:16:06 +00:00
#[tracing::instrument(level = "debug", skip(self, revision))]
2022-01-25 12:37:48 +00:00
pub async fn add_local_revision<C>(&self, revision: &Revision) -> Result<(), FlowyError>
where
C: RevisionCompact,
{
2022-01-01 15:09:13 +00:00
if revision.delta_data.is_empty() {
return Err(FlowyError::internal().context("Delta data should be empty"));
}
2022-01-26 15:29:18 +00:00
let mut write_guard = self.cache.write().await;
let rev_id = write_guard.write_sync_revision::<C>(revision).await?;
self.rev_id_counter.set(rev_id);
2021-10-07 12:46:29 +00:00
Ok(())
}
2022-01-01 08:16:06 +00:00
#[tracing::instrument(level = "debug", skip(self), err)]
2021-12-16 13:31:36 +00:00
pub async fn ack_revision(&self, rev_id: i64) -> Result<(), FlowyError> {
2022-01-25 12:37:48 +00:00
if self.cache.write().await.ack_revision(rev_id).await.is_ok() {
#[cfg(feature = "flowy_unit_test")]
let _ = self.revision_ack_notifier.send(rev_id);
}
Ok(())
}
2022-01-24 09:35:58 +00:00
pub fn rev_id(&self) -> i64 {
self.rev_id_counter.value()
}
2022-01-02 02:34:42 +00:00
pub fn next_rev_id_pair(&self) -> (i64, i64) {
let cur = self.rev_id_counter.value();
let next = self.rev_id_counter.next();
(cur, next)
}
2021-12-25 13:44:45 +00:00
pub async fn get_revisions_in_range(&self, range: RevisionRange) -> Result<Vec<Revision>, FlowyError> {
2022-01-26 15:29:18 +00:00
let revisions = self.cache.read().await.revisions_in_range(&range).await?;
2021-12-25 13:44:45 +00:00
Ok(revisions)
}
2021-12-08 13:51:06 +00:00
2022-01-25 12:37:48 +00:00
pub async fn next_sync_revision(&self) -> FlowyResult<Option<Revision>> {
Ok(self.cache.read().await.next_sync_revision().await?)
2022-01-01 08:16:06 +00:00
}
2021-12-18 10:35:45 +00:00
2021-12-25 13:44:45 +00:00
pub async fn get_revision(&self, rev_id: i64) -> Option<Revision> {
2022-01-25 12:37:48 +00:00
self.cache.read().await.get(rev_id).await.map(|record| record.revision)
2021-12-25 13:44:45 +00:00
}
}
2021-12-08 06:17:40 +00:00
2022-01-25 12:37:48 +00:00
#[cfg(feature = "flowy_unit_test")]
impl RevisionManager {
pub async fn revision_cache(&self) -> Arc<RevisionCache> {
self.cache.read().await.inner.clone()
}
pub fn revision_ack_receiver(&self) -> tokio::sync::broadcast::Receiver<i64> {
self.revision_ack_notifier.subscribe()
2022-01-01 08:16:06 +00:00
}
}
2022-01-25 12:37:48 +00:00
struct RevisionCacheCompact {
object_id: String,
2022-01-26 15:29:18 +00:00
user_id: String,
2022-01-25 12:37:48 +00:00
inner: Arc<RevisionCache>,
sync_seq: RevisionSyncSequence,
}
impl RevisionCacheCompact {
2022-01-26 15:29:18 +00:00
fn new(object_id: &str, user_id: &str, inner: Arc<RevisionCache>) -> Self {
2022-01-25 12:37:48 +00:00
let sync_seq = RevisionSyncSequence::new();
let object_id = object_id.to_owned();
2022-01-26 15:29:18 +00:00
let user_id = user_id.to_owned();
2022-01-25 12:37:48 +00:00
Self {
object_id,
2022-01-26 15:29:18 +00:00
user_id,
2022-01-25 12:37:48 +00:00
inner,
sync_seq,
}
}
async fn add_ack_revision(&self, revision: &Revision) -> FlowyResult<()> {
self.inner.add(revision.clone(), RevisionState::Ack, true).await
}
2022-01-26 15:29:18 +00:00
async fn add_sync_revision(&mut self, revision: &Revision) -> FlowyResult<()> {
self.inner.add(revision.clone(), RevisionState::Sync, false).await?;
self.sync_seq.add(revision.rev_id)?;
Ok(())
}
#[tracing::instrument(level = "trace", skip(self, revision), fields(rev_id, compact_range), err)]
async fn write_sync_revision<C>(&mut self, revision: &Revision) -> FlowyResult<i64>
2022-01-25 12:37:48 +00:00
where
C: RevisionCompact,
{
2022-01-26 15:29:18 +00:00
match self.sync_seq.compact() {
None => {
tracing::Span::current().record("rev_id", &revision.rev_id);
self.inner.add(revision.clone(), RevisionState::Sync, true).await?;
self.sync_seq.add(revision.rev_id)?;
Ok(revision.rev_id)
}
Some((range, mut compact_seq)) => {
tracing::Span::current().record("compact_range", &format!("{}", range).as_str());
let mut revisions = self.inner.revisions_in_range(&range).await?;
if range.to_rev_ids().len() != revisions.len() {
debug_assert_eq!(range.to_rev_ids().len(), revisions.len());
}
// append the new revision
revisions.push(revision.clone());
// compact multiple revisions into one
let compact_revision = C::compact_revisions(&self.user_id, &self.object_id, revisions)?;
let rev_id = compact_revision.rev_id;
tracing::Span::current().record("rev_id", &rev_id);
// insert new revision
compact_seq.push_back(rev_id);
// replace the revisions in range with compact revision
self.inner.compact(&range, compact_revision).await?;
debug_assert_eq!(self.sync_seq.len(), compact_seq.len());
self.sync_seq.reset(compact_seq);
Ok(rev_id)
}
}
2022-01-25 12:37:48 +00:00
}
async fn ack_revision(&mut self, rev_id: i64) -> FlowyResult<()> {
if self.sync_seq.ack(&rev_id).is_ok() {
self.inner.ack(rev_id).await;
}
Ok(())
}
async fn next_sync_revision(&self) -> FlowyResult<Option<Revision>> {
2022-01-26 15:29:18 +00:00
if cfg!(feature = "flowy_unit_test") {
match self.sync_seq.next_rev_id() {
None => Ok(None),
Some(rev_id) => Ok(self.inner.get(rev_id).await.map(|record| record.revision)),
}
} else {
Ok(None)
2022-01-25 12:37:48 +00:00
}
}
async fn reset(&self, revisions: Vec<Revision>) -> FlowyResult<()> {
self.inner.reset_with_revisions(&self.object_id, revisions).await?;
Ok(())
}
}
impl std::ops::Deref for RevisionCacheCompact {
type Target = Arc<RevisionCache>;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
#[derive(Default)]
struct RevisionSyncSequence(VecDeque<i64>);
2022-01-02 02:34:42 +00:00
impl RevisionSyncSequence {
2022-01-24 09:35:58 +00:00
fn new() -> Self {
RevisionSyncSequence::default()
}
2022-01-01 08:16:06 +00:00
2022-01-26 15:29:18 +00:00
fn add(&mut self, new_rev_id: i64) -> FlowyResult<()> {
2022-01-01 08:16:06 +00:00
// The last revision's rev_id must be greater than the new one.
2022-01-25 12:37:48 +00:00
if let Some(rev_id) = self.0.back() {
2022-01-25 11:45:41 +00:00
if *rev_id >= new_rev_id {
2022-01-12 09:08:50 +00:00
return Err(
FlowyError::internal().context(format!("The new revision's id must be greater than {}", rev_id))
);
2022-01-01 08:16:06 +00:00
}
}
2022-01-25 12:37:48 +00:00
self.0.push_back(new_rev_id);
2022-01-01 08:16:06 +00:00
Ok(())
}
2022-01-25 12:37:48 +00:00
fn ack(&mut self, rev_id: &i64) -> FlowyResult<()> {
let cur_rev_id = self.0.front().cloned();
2022-01-25 11:45:41 +00:00
if let Some(pop_rev_id) = cur_rev_id {
2022-01-01 08:16:06 +00:00
if &pop_rev_id != rev_id {
let desc = format!(
"The ack rev_id:{} is not equal to the current rev_id:{}",
rev_id, pop_rev_id
);
return Err(FlowyError::internal().context(desc));
}
2022-01-25 12:37:48 +00:00
let _ = self.0.pop_front();
2022-01-01 08:16:06 +00:00
}
Ok(())
}
2022-01-25 12:37:48 +00:00
fn next_rev_id(&self) -> Option<i64> {
self.0.front().cloned()
}
2022-01-26 15:29:18 +00:00
fn reset(&mut self, new_seq: VecDeque<i64>) {
self.0 = new_seq;
}
fn len(&self) -> usize {
self.0.len()
}
// Compact the rev_ids into one except the current synchronizing rev_id.
fn compact(&self) -> Option<(RevisionRange, VecDeque<i64>)> {
self.next_rev_id()?;
let mut new_seq = self.0.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))
2022-01-24 09:35:58 +00:00
}
2021-12-08 06:17:40 +00:00
}
2021-12-17 16:23:26 +00:00
struct RevisionLoader {
2022-01-14 07:23:21 +00:00
object_id: String,
2021-12-17 16:23:26 +00:00
user_id: String,
2022-01-14 07:23:21 +00:00
cloud: Arc<dyn RevisionCloudService>,
2022-01-25 12:37:48 +00:00
cache: Arc<RwLock<RevisionCacheCompact>>,
2021-12-17 16:23:26 +00:00
}
impl RevisionLoader {
2022-01-25 12:37:48 +00:00
async fn load<C>(&self) -> Result<(Vec<Revision>, i64), FlowyError>
where
C: RevisionCompact,
{
let records = self.cache.read().await.batch_get(&self.object_id)?;
2021-12-17 16:23:26 +00:00
let revisions: Vec<Revision>;
2022-01-23 14:33:47 +00:00
let mut rev_id = 0;
2021-12-17 16:23:26 +00:00
if records.is_empty() {
2022-01-14 07:23:21 +00:00
let remote_revisions = self.cloud.fetch_object(&self.user_id, &self.object_id).await?;
for revision in &remote_revisions {
2022-01-23 14:33:47 +00:00
rev_id = revision.rev_id;
2022-01-25 12:37:48 +00:00
let _ = self.cache.read().await.add_ack_revision(revision).await?;
2022-01-14 07:23:21 +00:00
}
revisions = remote_revisions;
2021-12-17 16:23:26 +00:00
} else {
2022-01-25 12:37:48 +00:00
for record in &records {
rev_id = record.revision.rev_id;
if record.state == RevisionState::Sync {
// Sync the records if their state is RevisionState::Sync.
2022-01-26 15:29:18 +00:00
let _ = self.cache.write().await.add_sync_revision(&record.revision).await?;
2022-01-23 14:33:47 +00:00
}
}
2021-12-17 16:23:26 +00:00
revisions = records.into_iter().map(|record| record.revision).collect::<_>();
}
2022-01-23 14:33:47 +00:00
if let Some(revision) = revisions.last() {
debug_assert_eq!(rev_id, revision.rev_id);
}
Ok((revisions, rev_id))
2021-12-17 16:23:26 +00:00
}
}