Commit e88ebe7d5a58
tarball2 files changed+29-11
@@ -1,7 +1,7 @@
11
use std::{22
collections::BTreeMap,33
fmt::Debug,4
- sync::{Arc, RwLock as StdRwLock},4
+ sync::{Arc, RwLock as StdRwLock, atomic::AtomicBool},55
time::Duration,66
};77
@@ -301,6 +301,7 @@ impl SlidingSyncBuilder {
301301
extensions: self.extensions.unwrap_or_default(),302302
303303
internal_channel: internal_channel_sender,304
+ event_cache_behind: AtomicBool::new(false),304305
305306
poll_timeout: self.poll_timeout,306307
network_timeout: self.network_timeout,@@ -26,7 +26,10 @@ use std::{
2626
collections::{BTreeMap, btree_map::Entry},2727
fmt::Debug,2828
future::Future,29
- sync::{Arc, RwLock as StdRwLock, RwLockWriteGuard as StdRwLockWriteGuard},29
+ sync::{30
+ Arc, RwLock as StdRwLock, RwLockWriteGuard as StdRwLockWriteGuard,31
+ atomic::{AtomicBool, Ordering},32
+ },3033
time::Duration,3134
};3235
@@ -53,10 +56,14 @@ pub use self::{builder::*, client::VersionBuilderError, error::*, list::*};
5356
use self::{cache::restore_sliding_sync_state, client::SlidingSyncResponseProcessor};5457
use crate::{Client, Result, config::RequestConfig};5558
56
-/// How long a rooms response waits for the event cache before its `pos` is57
-/// saved anyway.59
+/// How long a rooms response waits for the event cache before it gives up on60
+/// saving its `pos`.5861
const ROOM_UPDATES_WAIT: Duration = Duration::from_secs(10);5962
63
+/// The same wait once the event cache is known to be behind, long enough for a64
+/// recovered one to handle the latest response.65
+const ROOM_UPDATES_RECHECK: Duration = Duration::from_secs(1);66
+6067
/// The Sliding Sync instance.6168
///6269
/// It is OK to clone this type as much as you need: cloning it is cheap.@@ -123,6 +130,10 @@ pub(super) struct SlidingSyncInner {
123130
/// Internal channel used to pass messages between Sliding Sync and other124131
/// types.125132
internal_channel: Sender<SlidingSyncInternalMessage>,133
+134
+ /// Whether the event cache had not handled every response when the last135
+ /// one was handled, in which case `pos` was not saved.136
+ event_cache_behind: AtomicBool,126137
}127138
128139
impl SlidingSync {@@ -654,15 +665,21 @@ impl SlidingSync {
654665
.await?;655666
656667
// A `pos` saved before the event cache stores this response's events657
- // resumes past them after a kill, and no gap is left to fetch them.658
- if this.must_process_rooms_response().await659
- && !this.inner.client.event_cache().wait_for_room_updates(ROOM_UPDATES_WAIT).await660
- {661
- warn!("Saving the position before the event cache handled the response");668
+ // resumes past them after a kill, and no gap is left to fetch them. The669
+ // event cache can stall for good, so a lagging one leaves the last saved670
+ // `pos` in place, and a restart replays from there instead.671
+ let stored = !this.must_process_rooms_response().await || {672
+ let behind = this.inner.event_cache_behind.load(Ordering::SeqCst);673
+ let wait = if behind { ROOM_UPDATES_RECHECK } else { ROOM_UPDATES_WAIT };674
+ this.inner.client.event_cache().wait_for_room_updates(wait).await675
+ };676
+ this.inner.event_cache_behind.store(!stored, Ordering::SeqCst);677
+ if stored {678
+ this.cache_to_storage(&position_guard).await?;679
+ } else {680
+ warn!("Not saving the position, the event cache hasn't handled every response");662681
}663682
664
- this.cache_to_storage(&position_guard).await?;665
-666683
// Release the position guard lock.667684
// It means that other responses can be generated and then handled later.668685
drop(position_guard);