2022-01-14 15:23:21 +08:00
|
|
|
mod disk;
|
|
|
|
mod memory;
|
|
|
|
|
|
|
|
use crate::cache::{
|
|
|
|
disk::{RevisionChangeset, RevisionDiskCache, RevisionTableState, SQLitePersistence},
|
|
|
|
memory::{RevisionMemoryCache, RevisionMemoryCacheDelegate},
|
2021-12-08 14:17:40 +08:00
|
|
|
};
|
2022-01-25 20:37:48 +08:00
|
|
|
use crate::RevisionCompact;
|
2022-01-01 14:23:58 +08:00
|
|
|
use flowy_collaboration::entities::revision::{Revision, RevisionRange, RevisionState};
|
2021-12-08 14:17:40 +08:00
|
|
|
use flowy_database::ConnectionPool;
|
2022-01-14 15:23:21 +08:00
|
|
|
use flowy_error::{internal_error, FlowyError, FlowyResult};
|
2022-01-04 15:05:52 +08:00
|
|
|
use std::{
|
|
|
|
borrow::Cow,
|
|
|
|
sync::{
|
|
|
|
atomic::{AtomicI64, Ordering::SeqCst},
|
|
|
|
Arc,
|
|
|
|
},
|
2021-12-19 21:10:50 +08:00
|
|
|
};
|
2022-01-01 16:53:12 +08:00
|
|
|
use tokio::task::spawn_blocking;
|
2022-01-25 20:37:48 +08:00
|
|
|
|
2022-01-22 18:48:43 +08:00
|
|
|
pub const REVISION_WRITE_INTERVAL_IN_MILLIS: u64 = 600;
|
2021-12-08 14:17:40 +08:00
|
|
|
|
2022-01-14 15:23:21 +08:00
|
|
|
pub struct RevisionCache {
|
2022-01-17 11:55:36 +08:00
|
|
|
object_id: String,
|
2022-01-14 15:23:21 +08:00
|
|
|
disk_cache: Arc<dyn RevisionDiskCache<Error = FlowyError>>,
|
|
|
|
memory_cache: Arc<RevisionMemoryCache>,
|
2021-12-18 18:35:45 +08:00
|
|
|
latest_rev_id: AtomicI64,
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
|
|
|
|
2022-01-19 16:00:11 +08:00
|
|
|
pub fn mk_revision_disk_cache(
|
|
|
|
user_id: &str,
|
|
|
|
pool: Arc<ConnectionPool>,
|
|
|
|
) -> Arc<dyn RevisionDiskCache<Error = FlowyError>> {
|
|
|
|
Arc::new(SQLitePersistence::new(user_id, pool))
|
|
|
|
}
|
|
|
|
|
2022-01-14 15:23:21 +08:00
|
|
|
impl RevisionCache {
|
2022-01-17 11:55:36 +08:00
|
|
|
pub fn new(user_id: &str, object_id: &str, pool: Arc<ConnectionPool>) -> RevisionCache {
|
2022-01-04 15:05:52 +08:00
|
|
|
let disk_cache = Arc::new(SQLitePersistence::new(user_id, pool));
|
2022-01-17 11:55:36 +08:00
|
|
|
let memory_cache = Arc::new(RevisionMemoryCache::new(object_id, Arc::new(disk_cache.clone())));
|
|
|
|
let object_id = object_id.to_owned();
|
2021-12-08 14:17:40 +08:00
|
|
|
Self {
|
2022-01-17 11:55:36 +08:00
|
|
|
object_id,
|
2021-12-18 00:23:26 +08:00
|
|
|
disk_cache,
|
2021-12-08 14:17:40 +08:00
|
|
|
memory_cache,
|
2021-12-18 18:35:45 +08:00
|
|
|
latest_rev_id: AtomicI64::new(0),
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-01-25 19:45:41 +08:00
|
|
|
pub async fn add(&self, revision: Revision, state: RevisionState, write_to_disk: bool) -> FlowyResult<()> {
|
2021-12-13 22:46:35 +08:00
|
|
|
if self.memory_cache.contains(&revision.rev_id) {
|
2022-01-07 17:37:11 +08:00
|
|
|
return Err(FlowyError::internal().context(format!("Duplicate revision: {} {:?}", revision.rev_id, state)));
|
2021-12-13 22:46:35 +08:00
|
|
|
}
|
2022-01-02 10:34:42 +08:00
|
|
|
let state = state.as_ref().clone();
|
2021-12-18 18:35:45 +08:00
|
|
|
let rev_id = revision.rev_id;
|
2022-01-01 16:53:12 +08:00
|
|
|
let record = RevisionRecord {
|
|
|
|
revision,
|
|
|
|
state,
|
|
|
|
write_to_disk,
|
|
|
|
};
|
2022-01-07 17:37:11 +08:00
|
|
|
|
2022-01-25 19:45:41 +08:00
|
|
|
self.memory_cache.add(Cow::Owned(record)).await;
|
2022-01-01 16:16:06 +08:00
|
|
|
self.set_latest_rev_id(rev_id);
|
2022-01-25 19:45:41 +08:00
|
|
|
Ok(())
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
|
|
|
|
2022-01-23 12:14:00 +08:00
|
|
|
pub async fn ack(&self, rev_id: i64) {
|
|
|
|
self.memory_cache.ack(&rev_id).await;
|
|
|
|
}
|
2021-12-09 19:01:58 +08:00
|
|
|
|
2022-01-01 16:16:06 +08:00
|
|
|
pub async fn get(&self, rev_id: i64) -> Option<RevisionRecord> {
|
|
|
|
match self.memory_cache.get(&rev_id).await {
|
2022-01-17 11:55:36 +08:00
|
|
|
None => match self
|
|
|
|
.disk_cache
|
|
|
|
.read_revision_records(&self.object_id, Some(vec![rev_id]))
|
|
|
|
{
|
2022-01-01 14:23:58 +08:00
|
|
|
Ok(mut records) => {
|
2022-01-25 19:45:41 +08:00
|
|
|
let record = records.pop()?;
|
|
|
|
assert!(records.is_empty());
|
|
|
|
Some(record)
|
2022-01-23 12:14:00 +08:00
|
|
|
}
|
2021-12-18 18:35:45 +08:00
|
|
|
Err(e) => {
|
|
|
|
tracing::error!("{}", e);
|
|
|
|
None
|
2022-01-23 12:14:00 +08:00
|
|
|
}
|
2021-12-18 18:35:45 +08:00
|
|
|
},
|
|
|
|
Some(revision) => Some(revision),
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-01-01 16:16:06 +08:00
|
|
|
pub fn batch_get(&self, doc_id: &str) -> FlowyResult<Vec<RevisionRecord>> {
|
|
|
|
self.disk_cache.read_revision_records(doc_id, None)
|
|
|
|
}
|
|
|
|
|
2022-01-25 20:37:48 +08:00
|
|
|
// Read the revision which rev_id >= range.start && rev_id <= range.end
|
2021-12-14 18:04:51 +08:00
|
|
|
pub async fn revisions_in_range(&self, range: RevisionRange) -> FlowyResult<Vec<Revision>> {
|
2022-01-01 16:16:06 +08:00
|
|
|
let mut records = self.memory_cache.get_with_range(&range).await?;
|
2021-12-18 18:35:45 +08:00
|
|
|
let range_len = range.len() as usize;
|
|
|
|
if records.len() != range_len {
|
|
|
|
let disk_cache = self.disk_cache.clone();
|
2022-01-25 20:37:48 +08:00
|
|
|
let object_id = self.object_id.clone();
|
|
|
|
records = spawn_blocking(move || disk_cache.read_revision_records_with_range(&object_id, &range))
|
2021-12-18 18:35:45 +08:00
|
|
|
.await
|
|
|
|
.map_err(internal_error)??;
|
|
|
|
|
|
|
|
if records.len() != range_len {
|
2022-01-14 15:23:21 +08:00
|
|
|
tracing::error!("Revisions len is not equal to range required");
|
2021-12-18 18:35:45 +08:00
|
|
|
}
|
|
|
|
}
|
2021-12-18 00:23:26 +08:00
|
|
|
Ok(records
|
|
|
|
.into_iter()
|
|
|
|
.map(|record| record.revision)
|
|
|
|
.collect::<Vec<Revision>>())
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
|
|
|
|
2022-01-25 20:37:48 +08:00
|
|
|
#[tracing::instrument(level = "debug", skip(self, revisions))]
|
|
|
|
pub async fn reset_with_revisions(&self, object_id: &str, revisions: Vec<Revision>) -> FlowyResult<()> {
|
2022-01-02 10:34:42 +08:00
|
|
|
let revision_records = revisions
|
|
|
|
.to_vec()
|
2022-01-01 16:16:06 +08:00
|
|
|
.into_iter()
|
|
|
|
.map(|revision| RevisionRecord {
|
|
|
|
revision,
|
2022-01-14 15:23:21 +08:00
|
|
|
state: RevisionState::Sync,
|
2022-01-02 10:34:42 +08:00
|
|
|
write_to_disk: false,
|
2022-01-01 16:16:06 +08:00
|
|
|
})
|
|
|
|
.collect::<Vec<_>>();
|
|
|
|
|
2022-01-02 10:34:42 +08:00
|
|
|
let _ = self.memory_cache.reset_with_revisions(&revision_records).await?;
|
2022-01-25 20:37:48 +08:00
|
|
|
let _ = self.disk_cache.reset_object(object_id, revision_records)?;
|
2022-01-02 10:34:42 +08:00
|
|
|
Ok(())
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
2022-01-01 16:16:06 +08:00
|
|
|
|
|
|
|
#[inline]
|
|
|
|
fn set_latest_rev_id(&self, rev_id: i64) {
|
|
|
|
let _ = self.latest_rev_id.fetch_update(SeqCst, SeqCst, |_e| Some(rev_id));
|
|
|
|
}
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
|
|
|
|
2022-01-04 15:05:52 +08:00
|
|
|
impl RevisionMemoryCacheDelegate for Arc<SQLitePersistence> {
|
2022-01-08 18:21:13 +08:00
|
|
|
#[tracing::instrument(level = "trace", skip(self, records), fields(checkpoint_result), err)]
|
2022-01-01 16:53:12 +08:00
|
|
|
fn checkpoint_tick(&self, mut records: Vec<RevisionRecord>) -> FlowyResult<()> {
|
2022-01-01 14:23:58 +08:00
|
|
|
let conn = &*self.pool.get().map_err(internal_error)?;
|
2022-01-01 16:53:12 +08:00
|
|
|
records.retain(|record| record.write_to_disk);
|
|
|
|
if !records.is_empty() {
|
2022-01-07 17:37:11 +08:00
|
|
|
tracing::Span::current().record(
|
|
|
|
"checkpoint_result",
|
|
|
|
&format!("{} records were saved", records.len()).as_str(),
|
|
|
|
);
|
2022-01-23 12:14:00 +08:00
|
|
|
let _ = self.write_revision_records(records, conn)?;
|
2022-01-01 16:53:12 +08:00
|
|
|
}
|
|
|
|
Ok(())
|
2022-01-01 14:23:58 +08:00
|
|
|
}
|
2021-12-08 14:17:40 +08:00
|
|
|
|
2022-01-14 15:23:21 +08:00
|
|
|
fn receive_ack(&self, object_id: &str, rev_id: i64) {
|
2022-01-01 14:23:58 +08:00
|
|
|
let changeset = RevisionChangeset {
|
2022-01-14 15:23:21 +08:00
|
|
|
object_id: object_id.to_string(),
|
2021-12-18 18:35:45 +08:00
|
|
|
rev_id: rev_id.into(),
|
2022-01-01 16:16:06 +08:00
|
|
|
state: RevisionTableState::Ack,
|
2021-12-18 18:35:45 +08:00
|
|
|
};
|
2022-01-01 16:16:06 +08:00
|
|
|
match self.update_revision_record(vec![changeset]) {
|
2022-01-23 12:14:00 +08:00
|
|
|
Ok(_) => {}
|
2021-12-18 18:35:45 +08:00
|
|
|
Err(e) => tracing::error!("{}", e),
|
|
|
|
}
|
2021-12-08 14:17:40 +08:00
|
|
|
}
|
|
|
|
}
|
2021-12-08 21:51:06 +08:00
|
|
|
|
2021-12-19 21:10:50 +08:00
|
|
|
#[derive(Clone)]
|
|
|
|
pub struct RevisionRecord {
|
|
|
|
pub revision: Revision,
|
2022-01-01 14:23:58 +08:00
|
|
|
pub state: RevisionState,
|
2022-01-01 16:53:12 +08:00
|
|
|
pub write_to_disk: bool,
|
2021-12-19 21:10:50 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
impl RevisionRecord {
|
2022-01-23 12:14:00 +08:00
|
|
|
pub fn ack(&mut self) {
|
|
|
|
self.state = RevisionState::Ack;
|
|
|
|
}
|
2021-12-19 21:10:50 +08:00
|
|
|
}
|