use std::{
collections::{BTreeMap, HashMap},
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use futures_util::{StreamExt, future::BoxFuture, stream};
use inline_sdk::proto;
use prost::Message as _;
use tokio::sync::{Mutex, Semaphore};
use crate::{
BackendError, BackendResult, ClientErrorCategory, ClientEvent, ClientEventDelivery,
ClientStore, InlineId, PendingSyncBatch, StoreError, SyncBucketKey, SyncBucketPeer,
SyncBucketState, SyncState,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct SyncConfig {
pub last_sync_safety_gap_seconds: i64,
pub max_concurrent_bucket_fetches: usize,
pub initial_lookback_seconds: i64,
pub stale_state_max_age_seconds: i64,
pub max_total_updates: i32,
pub page_limit: i32,
pub cold_start_total_limit: i32,
pub inconsistent_empty_retry_attempts: u32,
pub inconsistent_empty_retry_delay_ms: u64,
}
impl Default for SyncConfig {
fn default() -> Self {
Self {
last_sync_safety_gap_seconds: 15,
max_concurrent_bucket_fetches: 4,
initial_lookback_seconds: 5 * 24 * 60 * 60,
stale_state_max_age_seconds: 14 * 24 * 60 * 60,
max_total_updates: 1_000,
page_limit: 200,
cold_start_total_limit: 50,
inconsistent_empty_retry_attempts: 3,
inconsistent_empty_retry_delay_ms: 250,
}
}
}
pub(crate) trait SyncHost: Clone + Send + Sync + 'static {
fn get_updates_state(
&self,
date: i64,
) -> BoxFuture<'static, BackendResult<proto::GetUpdatesStateResult>>;
fn get_updates(
&self,
input: proto::GetUpdatesInput,
) -> BoxFuture<'static, BackendResult<proto::GetUpdatesResult>>;
fn apply_sync_batch(
&self,
updates: Vec<proto::Update>,
sidecars: Option<proto::UpdateSidecars>,
) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>>;
fn repair_bucket(
&self,
key: SyncBucketKey,
) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>>;
}
#[derive(Debug)]
pub(crate) struct SyncManager {
store: Arc<dyn ClientStore>,
config: SyncConfig,
bucket_locks: Mutex<HashMap<SyncBucketKey, Arc<Mutex<()>>>>,
fetch_limiter: Arc<Semaphore>,
sync_state_lock: Mutex<()>,
}
impl SyncManager {
pub(crate) fn new(store: Arc<dyn ClientStore>, config: SyncConfig) -> Self {
let max_fetches = config.max_concurrent_bucket_fetches.max(1);
Self {
store,
config,
bucket_locks: Mutex::new(HashMap::new()),
fetch_limiter: Arc::new(Semaphore::new(max_fetches)),
sync_state_lock: Mutex::new(()),
}
}
pub(crate) async fn discover<H: SyncHost>(
&self,
host: &H,
) -> BackendResult<Vec<ClientEventDelivery>> {
let mut events = self.recover_pending_batches(host).await?;
let state = self.prepared_sync_state().await?;
log::debug!(
"starting Inline update discovery from date={}",
state.last_sync_date
);
let result = host.get_updates_state(state.last_sync_date).await?;
if result.updates_found == Some(false) {
self.update_last_sync_date(result.date).await?;
}
events.extend(
self.process_bucket(host, SyncBucketKey::User, Vec::new(), None, true)
.await?,
);
log::debug!(
"finished Inline update discovery date={} events={}",
result.date,
events.len()
);
Ok(events)
}
pub(crate) async fn process_realtime<H: SyncHost>(
&self,
host: &H,
updates: Vec<proto::Update>,
) -> BackendResult<Vec<ClientEventDelivery>> {
let mut events = self.recover_pending_batches(host).await?;
let received_count = updates.len();
let mut direct = Vec::new();
let mut buckets = HashMap::<SyncBucketKey, Vec<proto::Update>>::new();
let mut targets = HashMap::<SyncBucketKey, i64>::new();
for update in updates {
match update.update.as_ref() {
Some(proto::update::Update::ChatHasNewUpdates(hint)) => {
if let Some(key) = chat_hint_bucket_key(hint) {
targets
.entry(key)
.and_modify(|seq| *seq = (*seq).max(i64::from(hint.update_seq)))
.or_insert(i64::from(hint.update_seq));
}
}
Some(proto::update::Update::SpaceHasNewUpdates(hint)) => {
let key = SyncBucketKey::Space {
space_id: InlineId::new(hint.space_id),
};
targets
.entry(key)
.and_modify(|seq| *seq = (*seq).max(i64::from(hint.update_seq)))
.or_insert(i64::from(hint.update_seq));
}
_ => {
let seq = update_seq(&update);
if seq > 0
&& let Some(key) = bucket_key_for_update(&update)
{
buckets.entry(key).or_default().push(update);
} else {
direct.push(update);
}
}
}
}
if !direct.is_empty() {
let max_date = max_update_date(&direct);
let applied = host.apply_sync_batch(direct, None).await?;
let applied = self
.store
.append_client_events(applied)
.await
.map_err(store_error_to_backend)?;
self.update_last_sync_date(max_date).await?;
events.extend(applied);
}
for key in targets.keys() {
buckets.entry(*key).or_default();
}
let mut ordered = buckets
.into_iter()
.map(|(key, updates)| {
let target = targets.get(&key).copied();
(key, updates, target)
})
.collect::<Vec<_>>();
let bucket_count = ordered.len();
ordered.sort_by_key(|(key, _, _)| bucket_sort_key(*key));
if let Some(user_index) = ordered
.iter()
.position(|(key, _, _)| *key == SyncBucketKey::User)
{
let (key, updates, target) = ordered.remove(user_index);
events.extend(
self.process_bucket(host, key, updates, target, false)
.await?,
);
}
let concurrency = self.config.max_concurrent_bucket_fetches.max(1);
let mut results = stream::iter(ordered)
.map(|(key, updates, target)| async move {
(
bucket_sort_key(key),
self.process_bucket(host, key, updates, target, false).await,
)
})
.buffer_unordered(concurrency)
.collect::<Vec<_>>()
.await;
results.sort_by_key(|(sort_key, _)| *sort_key);
for (_, result) in results {
events.extend(result?);
}
log::debug!(
"processed Inline realtime batch received={received_count} buckets={bucket_count} events={}",
events.len()
);
Ok(events)
}
async fn process_bucket<H: SyncHost>(
&self,
host: &H,
key: SyncBucketKey,
realtime_updates: Vec<proto::Update>,
target_seq: Option<i64>,
force_fetch: bool,
) -> BackendResult<Vec<ClientEventDelivery>> {
let bucket_lock = self.bucket_lock(key).await;
let _bucket_guard = bucket_lock.lock().await;
let mut state = self
.store
.sync_bucket_state(key)
.await
.map_err(store_error_to_backend)?;
let mut buffered = BTreeMap::new();
for update in realtime_updates {
let seq = update_seq(&update);
if seq > state.seq {
insert_unique_update(&mut buffered, seq, update, "realtime buffer")?;
}
}
let mut events = self
.drain_contiguous(host, key, &mut state, &mut buffered)
.await?;
let target_seq = target_seq
.filter(|target| *target > state.seq)
.or_else(|| buffered.last_key_value().map(|(seq, _)| *seq));
if !force_fetch && target_seq.is_none() && buffered.is_empty() && state.seq > 0 {
return Ok(events);
}
events.extend(
self.fetch_bucket(host, key, state, buffered, target_seq)
.await?,
);
Ok(events)
}
async fn drain_contiguous<H: SyncHost>(
&self,
host: &H,
key: SyncBucketKey,
state: &mut SyncBucketState,
buffered: &mut BTreeMap<i64, proto::Update>,
) -> BackendResult<Vec<ClientEventDelivery>> {
let mut updates = Vec::new();
let mut next_seq = state.seq + 1;
while let Some(update) = buffered.remove(&next_seq) {
updates.push(update);
next_seq += 1;
}
if updates.is_empty() {
return Ok(Vec::new());
}
let committed = SyncBucketState {
seq: next_seq - 1,
date: state.date.max(max_update_date(&updates)),
};
let events = self
.commit_sync_batch(host, key, committed, updates, None)
.await?;
self.update_last_sync_date(committed.date).await?;
*state = committed;
Ok(events)
}
async fn fetch_bucket<H: SyncHost>(
&self,
host: &H,
key: SyncBucketKey,
initial: SyncBucketState,
buffered: BTreeMap<i64, proto::Update>,
target_seq: Option<i64>,
) -> BackendResult<Vec<ClientEventDelivery>> {
let _permit = self
.fetch_limiter
.clone()
.acquire_owned()
.await
.map_err(|_| BackendError::new(ClientErrorCategory::Internal, "sync fetch stopped"))?;
let mut cold_start = initial.seq == 0 || initial.date == 0;
let mut committed_floor = initial.seq;
let mut current_seq = initial.seq;
let mut final_date = initial.date;
let hard_end = target_seq.or_else(|| buffered.last_key_value().map(|(seq, _)| *seq));
let mut buffered = buffered;
let mut slice_end = None;
let mut fetched = BTreeMap::<i64, proto::Update>::new();
let mut sidecars = proto::UpdateSidecars::default();
let mut page_count = 0_u64;
let mut inconsistent_empty_attempts = 0_u32;
let mut deliveries = Vec::new();
log::debug!(
"starting Inline bucket fetch kind={} start_seq={} target_seq={:?} cold_start={cold_start}",
bucket_kind(key),
current_seq,
hard_end
);
loop {
page_count += 1;
let request_end = slice_end.or(hard_end).filter(|end| *end > current_seq);
let response = host
.get_updates(proto::GetUpdatesInput {
bucket: Some(protocol_bucket(key)),
start_seq: current_seq,
total_limit: if cold_start && slice_end.is_none() {
self.config.cold_start_total_limit
} else {
self.config.max_total_updates
},
seq_end: request_end.unwrap_or_default(),
limit: self.config.page_limit,
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
})
.await?;
let result_type = proto::get_updates_result::ResultType::try_from(response.result_type)
.unwrap_or(proto::get_updates_result::ResultType::Unspecified);
if result_type == proto::get_updates_result::ResultType::TooLong {
if response.seq <= current_seq {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync received a non-advancing TOO_LONG pointer",
));
}
if cold_start {
let target_seq = response.seq;
let target_date = final_date.max(response.date);
log::warn!(
"repairing cold Inline bucket after TOO_LONG kind={} target_seq={target_seq}",
bucket_kind(key)
);
deliveries.extend(
self.repair_cold_bucket(host, key, target_seq, target_date)
.await?,
);
cold_start = false;
committed_floor = target_seq;
current_seq = target_seq;
final_date = target_date;
fetched.clear();
sidecars = proto::UpdateSidecars::default();
slice_end = None;
buffered.retain(|seq, _| *seq > target_seq);
if hard_end.is_none_or(|end| target_seq >= end) {
return Ok(deliveries);
}
continue;
}
let max_slice = current_seq + i64::from(self.config.max_total_updates);
slice_end = Some(response.seq.min(max_slice));
continue;
}
if response.seq < current_seq {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync server cursor moved backwards",
));
}
let previous_seq = current_seq;
let accounting = validate_response_page(previous_seq, &response)?;
current_seq = response.seq;
final_date = final_date.max(response.date);
let skipped_sequences = response
.skipped_sequences
.iter()
.map(|skipped| skipped.seq)
.collect::<std::collections::BTreeSet<_>>();
let page_sidecars = response.sidecars;
let page_updates = response.updates;
let empty = result_type == proto::get_updates_result::ResultType::Empty;
if accounting.requires_snapshot_repair {
log::warn!(
"repairing Inline bucket after server-classified snapshot gap kind={} target_seq={current_seq}",
bucket_kind(key)
);
deliveries.extend(
self.repair_cold_bucket(host, key, current_seq, final_date)
.await?,
);
return Ok(deliveries);
}
if cold_start {
if let Some(page_sidecars) = page_sidecars {
merge_sidecars(&mut sidecars, page_sidecars);
}
for update in page_updates {
let seq = update_seq(&update);
if seq > committed_floor {
insert_unique_update(&mut fetched, seq, update, "fetched page")?;
}
}
} else if current_seq > previous_seq {
let mut page = BTreeMap::<i64, proto::Update>::new();
for update in page_updates {
let seq = update_seq(&update);
insert_unique_update(&mut page, seq, update, "fetched page")?;
}
let buffered_sequences = buffered
.range(..=current_seq)
.map(|(seq, _)| *seq)
.collect::<Vec<_>>();
for seq in &buffered_sequences {
if skipped_sequences.contains(seq) {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"realtime update conflicted with a server-classified skipped sequence",
));
}
if let Some(update) = buffered.get(seq).cloned() {
insert_unique_update(&mut page, *seq, update, "fetched/realtime merge")?;
}
}
let updates = page.into_values().collect::<Vec<_>>();
let committed = SyncBucketState {
seq: current_seq,
date: final_date.max(max_update_date(&updates)),
};
let has_sidecars = page_sidecars.as_ref().is_some_and(has_sidecars);
let page_deliveries = self
.commit_sync_batch(
host,
key,
committed,
updates,
has_sidecars.then_some(page_sidecars).flatten(),
)
.await?;
self.update_last_sync_date(committed.date).await?;
deliveries.extend(page_deliveries);
for seq in buffered_sequences {
buffered.remove(&seq);
}
committed_floor = committed.seq;
}
let final_page = response.r#final.unwrap_or(false) || empty;
if final_page && hard_end.is_some_and(|target| current_seq < target) {
if current_seq > previous_seq {
inconsistent_empty_attempts = 0;
}
inconsistent_empty_attempts += 1;
if inconsistent_empty_attempts <= self.config.inconsistent_empty_retry_attempts {
log::warn!(
"retrying Inline final page behind hint kind={} server_seq={} target_seq={:?} attempt={}",
bucket_kind(key),
current_seq,
hard_end,
inconsistent_empty_attempts
);
if self.config.inconsistent_empty_retry_delay_ms > 0 {
tokio::time::sleep(Duration::from_millis(
self.config.inconsistent_empty_retry_delay_ms,
))
.await;
}
continue;
}
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync final response remained behind the requested target",
));
}
inconsistent_empty_attempts = 0;
if current_seq == previous_seq && !final_page {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync received a non-progress response",
));
}
if let Some(end) = slice_end
&& current_seq >= end
{
slice_end = None;
if hard_end.is_some_and(|target| current_seq < target) {
continue;
}
}
if hard_end.is_some_and(|target| current_seq >= target) || final_page {
break;
}
}
if !cold_start {
log::debug!(
"finished incremental Inline bucket fetch kind={} pages={page_count} end_seq={} events={}",
bucket_kind(key),
committed_floor,
deliveries.len()
);
return Ok(deliveries);
}
for (seq, update) in buffered {
if seq > committed_floor && seq <= current_seq {
insert_unique_update(&mut fetched, seq, update, "fetched/realtime merge")?;
}
}
let updates = fetched.into_values().collect::<Vec<_>>();
let max_delivered_seq = updates.iter().map(update_seq).max().unwrap_or(initial.seq);
if current_seq > max_delivered_seq {
log::warn!(
"advancing cold Inline bucket pointer with explicit sequence accounting kind={} delivered_seq={max_delivered_seq} pointer_seq={current_seq}",
bucket_kind(key)
);
}
let committed = SyncBucketState {
seq: current_seq,
date: final_date.max(max_update_date(&updates)),
};
if cold_start {
let events = self
.repair_cold_bucket(host, key, committed.seq, committed.date)
.await?;
log::debug!(
"finished cold Inline bucket snapshot kind={} pages={page_count} end_seq={} events={}",
bucket_kind(key),
committed.seq,
events.len()
);
deliveries.extend(events);
return Ok(deliveries);
}
let has_sidecars = has_sidecars(&sidecars);
let update_count = updates.len();
let events = self
.commit_sync_batch(
host,
key,
committed,
updates,
has_sidecars.then_some(sidecars),
)
.await?;
self.update_last_sync_date(committed.date).await?;
log::debug!(
"finished Inline bucket fetch kind={} pages={page_count} end_seq={} updates={} events={}",
bucket_kind(key),
committed.seq,
update_count,
events.len()
);
deliveries.extend(events);
Ok(deliveries)
}
async fn repair_cold_bucket<H: SyncHost>(
&self,
host: &H,
key: SyncBucketKey,
target_seq: i64,
target_date: i64,
) -> BackendResult<Vec<ClientEventDelivery>> {
let events = host.repair_bucket(key).await?;
self.commit_cold_pointer(key, target_seq, target_date, events)
.await
}
async fn commit_cold_pointer(
&self,
key: SyncBucketKey,
target_seq: i64,
target_date: i64,
events: Vec<ClientEvent>,
) -> BackendResult<Vec<ClientEventDelivery>> {
let committed = SyncBucketState {
seq: target_seq,
date: target_date,
};
let deliveries = self
.store
.commit_pending_sync_batch_with_events(key, committed, events)
.await
.map_err(store_error_to_backend)?;
self.update_last_sync_date(committed.date).await?;
Ok(deliveries)
}
async fn commit_sync_batch<H: SyncHost>(
&self,
host: &H,
key: SyncBucketKey,
committed_state: SyncBucketState,
updates: Vec<proto::Update>,
sidecars: Option<proto::UpdateSidecars>,
) -> BackendResult<Vec<ClientEventDelivery>> {
validate_journal_updates(&updates)?;
let payload = encode_pending_payload(&updates, sidecars.as_ref())?;
self.store
.save_pending_sync_batch(PendingSyncBatch {
key,
committed_state,
payload,
})
.await
.map_err(store_error_to_backend)?;
let events = match host.apply_sync_batch(updates, sidecars).await {
Ok(events) => events,
Err(error) => {
if is_permanent_sync_error(&error) {
self.store
.discard_pending_sync_batch(key)
.await
.map_err(store_error_to_backend)?;
}
return Err(error);
}
};
let deliveries = self
.store
.commit_pending_sync_batch_with_events(key, committed_state, events)
.await
.map_err(store_error_to_backend)?;
Ok(deliveries)
}
async fn recover_pending_batches<H: SyncHost>(
&self,
host: &H,
) -> BackendResult<Vec<ClientEventDelivery>> {
let batches = self
.store
.pending_sync_batches()
.await
.map_err(store_error_to_backend)?;
let mut events = Vec::new();
let mut refetch = BTreeMap::<(u8, i64), (SyncBucketKey, i64)>::new();
for batch in batches {
let decoded = decode_pending_payload(&batch.payload).and_then(|(updates, sidecars)| {
validate_journal_updates(&updates)?;
Ok((updates, sidecars))
});
let (updates, sidecars) = match decoded {
Ok(decoded) => decoded,
Err(error) if is_permanent_sync_error(&error) => {
log::warn!(
"discarding incompatible Inline sync journal kind={} target_seq={}",
bucket_kind(batch.key),
batch.committed_state.seq
);
self.store
.discard_pending_sync_batch(batch.key)
.await
.map_err(store_error_to_backend)?;
refetch.insert(
bucket_sort_key(batch.key),
(batch.key, batch.committed_state.seq),
);
continue;
}
Err(error) => return Err(error),
};
let applied = match host.apply_sync_batch(updates, sidecars).await {
Ok(applied) => applied,
Err(error) if is_permanent_sync_error(&error) => {
log::warn!(
"discarding permanently failing Inline sync journal kind={} target_seq={}",
bucket_kind(batch.key),
batch.committed_state.seq
);
self.store
.discard_pending_sync_batch(batch.key)
.await
.map_err(store_error_to_backend)?;
refetch.insert(
bucket_sort_key(batch.key),
(batch.key, batch.committed_state.seq),
);
continue;
}
Err(error) => return Err(error),
};
events.extend(
self.store
.commit_pending_sync_batch_with_events(
batch.key,
batch.committed_state,
applied,
)
.await
.map_err(store_error_to_backend)?,
);
self.update_last_sync_date(batch.committed_state.date)
.await?;
}
for (_, (key, target_seq)) in refetch {
events.extend(
self.process_bucket(host, key, Vec::new(), Some(target_seq), true)
.await?,
);
}
Ok(events)
}
async fn bucket_lock(&self, key: SyncBucketKey) -> Arc<Mutex<()>> {
let mut locks = self.bucket_locks.lock().await;
locks
.entry(key)
.or_insert_with(|| Arc::new(Mutex::new(())))
.clone()
}
async fn prepared_sync_state(&self) -> BackendResult<SyncState> {
let mut state = self
.store
.sync_state()
.await
.map_err(store_error_to_backend)?;
let now = now_seconds();
if state.last_sync_date == 0
|| now.saturating_sub(state.last_sync_date) > self.config.stale_state_max_age_seconds
{
state.last_sync_date = now.saturating_sub(self.config.initial_lookback_seconds);
self.store
.save_sync_state(state)
.await
.map_err(store_error_to_backend)?;
}
Ok(state)
}
async fn update_last_sync_date(&self, max_applied_date: i64) -> BackendResult<()> {
if max_applied_date <= 0 {
return Ok(());
}
let _guard = self.sync_state_lock.lock().await;
let current = self
.store
.sync_state()
.await
.map_err(store_error_to_backend)?;
let proposed = max_applied_date.saturating_sub(self.config.last_sync_safety_gap_seconds);
if proposed <= current.last_sync_date {
return Ok(());
}
self.store
.save_sync_state(SyncState {
last_sync_date: proposed,
})
.await
.map_err(store_error_to_backend)
}
}
fn bucket_key_for_update(update: &proto::Update) -> Option<SyncBucketKey> {
use proto::update::Update;
match update.update.as_ref()? {
Update::NewMessage(value) => value
.message
.as_ref()?
.peer_id
.as_ref()
.and_then(chat_bucket),
Update::EditMessage(value) => value
.message
.as_ref()?
.peer_id
.as_ref()
.and_then(chat_bucket),
Update::DeleteMessages(value) => value.peer_id.as_ref().and_then(chat_bucket),
Update::ClearChatHistory(value) => match value.target.as_ref()? {
proto::update_clear_chat_history::Target::PeerId(peer) => chat_bucket(peer),
proto::update_clear_chat_history::Target::SpaceId(space_id) => {
Some(SyncBucketKey::Space {
space_id: InlineId::new(*space_id),
})
}
},
Update::MessageAttachment(value) => value.peer_id.as_ref().and_then(chat_bucket),
Update::UpdateReaction(value) => value
.reaction
.as_ref()
.map(|reaction| chat_key(reaction.chat_id)),
Update::DeleteReaction(value) => Some(chat_key(value.chat_id)),
Update::DeleteChat(value) => value.peer_id.as_ref().and_then(chat_bucket),
Update::MarkAsUnread(value) => value.peer_id.as_ref().and_then(chat_bucket),
Update::SpaceMemberAdd(value) => value
.member
.as_ref()
.map(|member| space_key(member.space_id)),
Update::SpaceMemberDelete(value) => Some(space_key(value.space_id)),
Update::SpaceMemberUpdate(value) => value
.member
.as_ref()
.map(|member| space_key(member.space_id)),
Update::JoinSpace(_)
| Update::UpdateUserStatus(_)
| Update::UpdateUserSettings(_)
| Update::UpdatedUser(_)
| Update::DialogArchived(_)
| Update::DialogNotificationSettings(_)
| Update::DialogFollowMode(_)
| Update::UpdateReadMaxId(_)
| Update::ChatOpen(_) => Some(SyncBucketKey::User),
Update::NewChat(value) => value.chat.as_ref()?.peer_id.as_ref().and_then(chat_bucket),
Update::ChatMoved(value) => value.chat.as_ref()?.peer_id.as_ref().and_then(chat_bucket),
Update::ParticipantAdd(value) => Some(chat_key(value.chat_id)),
Update::ParticipantDelete(value) => Some(chat_key(value.chat_id)),
Update::ParticipantGroupAdd(value) => Some(chat_key(value.chat_id)),
Update::ParticipantGroupDelete(value) => Some(chat_key(value.chat_id)),
Update::ChatVisibility(value) => Some(chat_key(value.chat_id)),
Update::ChatInfo(value) => Some(chat_key(value.chat_id)),
Update::PinnedMessages(value) => value.peer_id.as_ref().and_then(chat_bucket),
Update::ChatSkipPts(value) => Some(chat_key(value.chat_id)),
Update::SpaceSettings(value) => Some(space_key(value.space_id)),
_ => None,
}
}
fn chat_hint_bucket_key(hint: &proto::UpdateChatHasNewUpdates) -> Option<SyncBucketKey> {
hint.peer_id
.as_ref()
.and_then(chat_bucket)
.or_else(|| (hint.chat_id != 0).then(|| chat_key(hint.chat_id)))
}
fn chat_bucket(peer: &proto::Peer) -> Option<SyncBucketKey> {
let peer = match peer.r#type.as_ref()? {
proto::peer::Type::User(user) => SyncBucketPeer::User {
user_id: InlineId::new(user.user_id),
},
proto::peer::Type::Chat(chat) => SyncBucketPeer::Chat {
chat_id: InlineId::new(chat.chat_id),
},
};
Some(SyncBucketKey::Chat { peer })
}
fn chat_key(chat_id: i64) -> SyncBucketKey {
SyncBucketKey::Chat {
peer: SyncBucketPeer::Chat {
chat_id: InlineId::new(chat_id),
},
}
}
fn space_key(space_id: i64) -> SyncBucketKey {
SyncBucketKey::Space {
space_id: InlineId::new(space_id),
}
}
fn protocol_bucket(key: SyncBucketKey) -> proto::UpdateBucket {
let r#type = match key {
SyncBucketKey::User => proto::update_bucket::Type::User(proto::UpdateBucketUser {}),
SyncBucketKey::Space { space_id } => {
proto::update_bucket::Type::Space(proto::UpdateBucketSpace {
space_id: space_id.get(),
})
}
SyncBucketKey::Chat { peer } => {
let r#type = match peer {
SyncBucketPeer::User { user_id } => {
proto::input_peer::Type::User(proto::InputPeerUser {
user_id: user_id.get(),
})
}
SyncBucketPeer::Chat { chat_id } => {
proto::input_peer::Type::Chat(proto::InputPeerChat {
chat_id: chat_id.get(),
})
}
};
proto::update_bucket::Type::Chat(proto::UpdateBucketChat {
peer_id: Some(proto::InputPeer {
r#type: Some(r#type),
}),
})
}
};
proto::UpdateBucket {
r#type: Some(r#type),
}
}
fn update_seq(update: &proto::Update) -> i64 {
i64::from(update.seq.unwrap_or_default())
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
struct PageAccounting {
requires_snapshot_repair: bool,
}
fn validate_response_page(
previous_seq: i64,
response: &proto::GetUpdatesResult,
) -> BackendResult<PageAccounting> {
use proto::get_updates_result::ResultType;
use proto::sync_skipped_sequence::Reason;
let result_type = ResultType::try_from(response.result_type).map_err(|_| {
BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync response used an unknown result type",
)
})?;
if !matches!(result_type, ResultType::Slice | ResultType::Empty) {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync response used an invalid page result type",
));
}
if response.seq < previous_seq {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page cursor moved backwards",
));
}
let mut accounted = BTreeMap::<i64, &'static str>::new();
for update in &response.updates {
let seq = update_seq(update);
validate_page_sequence(previous_seq, response.seq, seq)?;
if accounted.insert(seq, "update").is_some() {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page contained a duplicate sequence",
));
}
}
let mut requires_snapshot_repair = false;
for skipped in &response.skipped_sequences {
validate_page_sequence(previous_seq, response.seq, skipped.seq)?;
let reason = Reason::try_from(skipped.reason).map_err(|_| {
BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page used an unknown skipped-sequence reason",
)
})?;
match reason {
Reason::IrrelevantToBucket => {}
Reason::SnapshotRepairRequired => requires_snapshot_repair = true,
Reason::Unspecified => {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page used an unspecified skipped-sequence reason",
));
}
}
if accounted.insert(skipped.seq, "skip").is_some() {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page accounted for one sequence more than once",
));
}
}
let delta = response.seq.saturating_sub(previous_seq);
let accounted_count = i64::try_from(accounted.len()).map_err(|_| {
BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page sequence count overflowed",
)
})?;
if accounted_count != delta {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page did not account for every advanced sequence",
));
}
Ok(PageAccounting {
requires_snapshot_repair,
})
}
fn validate_page_sequence(previous_seq: i64, response_seq: i64, seq: i64) -> BackendResult<()> {
if seq <= previous_seq || seq > response_seq {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"bucket sync page contained a sequence outside its cursor envelope",
));
}
Ok(())
}
fn insert_unique_update(
updates: &mut BTreeMap<i64, proto::Update>,
seq: i64,
update: proto::Update,
source: &'static str,
) -> BackendResult<()> {
if let Some(existing) = updates.get(&seq) {
if existing == &update {
return Ok(());
}
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
format!("conflicting Inline updates shared sequence {seq} in {source}"),
));
}
updates.insert(seq, update);
Ok(())
}
fn validate_journal_updates(updates: &[proto::Update]) -> BackendResult<()> {
for update in updates {
if update_seq(update) <= 0 {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"lossless sync journal contained an unsequenced update",
));
}
match update.update.as_ref() {
Some(proto::update::Update::ChatHasNewUpdates(_))
| Some(proto::update::Update::SpaceHasNewUpdates(_)) => {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"lossless sync journal contained a realtime hint",
));
}
Some(_) => {}
None => {
return Err(BackendError::new(
ClientErrorCategory::ProtocolMismatch,
"lossless sync journal contained an unknown or missing update payload",
));
}
}
}
Ok(())
}
fn is_permanent_sync_error(error: &BackendError) -> bool {
matches!(
error.category,
ClientErrorCategory::ProtocolMismatch | ClientErrorCategory::Unsupported
)
}
fn max_update_date(updates: &[proto::Update]) -> i64 {
updates
.iter()
.filter_map(|update| update.date)
.max()
.unwrap_or_default()
}
fn merge_sidecars(target: &mut proto::UpdateSidecars, page: proto::UpdateSidecars) {
target.users.extend(page.users);
target.chats.extend(page.chats);
target.dialogs.extend(page.dialogs);
target.spaces.extend(page.spaces);
target.user_groups.extend(page.user_groups);
}
fn has_sidecars(sidecars: &proto::UpdateSidecars) -> bool {
!sidecars.users.is_empty()
|| !sidecars.chats.is_empty()
|| !sidecars.dialogs.is_empty()
|| !sidecars.spaces.is_empty()
|| !sidecars.user_groups.is_empty()
}
fn bucket_sort_key(key: SyncBucketKey) -> (u8, i64) {
match key {
SyncBucketKey::User => (0, 0),
SyncBucketKey::Space { space_id } => (1, space_id.get()),
SyncBucketKey::Chat {
peer: SyncBucketPeer::User { user_id },
} => (2, user_id.get()),
SyncBucketKey::Chat {
peer: SyncBucketPeer::Chat { chat_id },
} => (3, chat_id.get()),
}
}
fn bucket_kind(key: SyncBucketKey) -> &'static str {
match key {
SyncBucketKey::User => "user",
SyncBucketKey::Space { .. } => "space",
SyncBucketKey::Chat { .. } => "chat",
}
}
fn store_error_to_backend(error: StoreError) -> BackendError {
BackendError::new(error.category, error.message)
}
fn now_seconds() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs() as i64
}
const PENDING_SYNC_PAYLOAD_MAGIC: &[u8; 4] = b"ISB1";
const MAX_PENDING_SYNC_ITEMS: usize = 100_000;
const MAX_PENDING_SYNC_ITEM_BYTES: usize = 64 * 1024 * 1024;
fn encode_pending_payload(
updates: &[proto::Update],
sidecars: Option<&proto::UpdateSidecars>,
) -> BackendResult<Vec<u8>> {
if updates.len() > MAX_PENDING_SYNC_ITEMS {
return Err(BackendError::new(
ClientErrorCategory::Internal,
"sync batch is too large to journal",
));
}
let mut payload = Vec::new();
payload.extend_from_slice(PENDING_SYNC_PAYLOAD_MAGIC);
append_u32(&mut payload, updates.len())?;
for update in updates {
let encoded = update.encode_to_vec();
append_bytes(&mut payload, &encoded)?;
}
match sidecars {
Some(sidecars) => {
payload.push(1);
append_bytes(&mut payload, &sidecars.encode_to_vec())?;
}
None => payload.push(0),
}
Ok(payload)
}
fn decode_pending_payload(
payload: &[u8],
) -> BackendResult<(Vec<proto::Update>, Option<proto::UpdateSidecars>)> {
if !payload.starts_with(PENDING_SYNC_PAYLOAD_MAGIC) {
return Err(invalid_pending_payload("unsupported journal version"));
}
let mut offset = PENDING_SYNC_PAYLOAD_MAGIC.len();
let count = read_u32(payload, &mut offset)?;
if count > MAX_PENDING_SYNC_ITEMS {
return Err(invalid_pending_payload("update count exceeds safety limit"));
}
let mut updates = Vec::with_capacity(count);
for _ in 0..count {
let bytes = read_bytes(payload, &mut offset)?;
updates.push(
proto::Update::decode(bytes)
.map_err(|_| invalid_pending_payload("update protobuf is invalid"))?,
);
}
let sidecars = match payload.get(offset).copied() {
Some(0) => {
offset += 1;
None
}
Some(1) => {
offset += 1;
let bytes = read_bytes(payload, &mut offset)?;
Some(
proto::UpdateSidecars::decode(bytes)
.map_err(|_| invalid_pending_payload("sidecars protobuf is invalid"))?,
)
}
_ => return Err(invalid_pending_payload("sidecars flag is invalid")),
};
if offset != payload.len() {
return Err(invalid_pending_payload("journal has trailing bytes"));
}
Ok((updates, sidecars))
}
fn append_u32(payload: &mut Vec<u8>, value: usize) -> BackendResult<()> {
let value = u32::try_from(value).map_err(|_| {
BackendError::new(
ClientErrorCategory::Internal,
"sync journal length exceeds u32",
)
})?;
payload.extend_from_slice(&value.to_le_bytes());
Ok(())
}
fn append_bytes(payload: &mut Vec<u8>, bytes: &[u8]) -> BackendResult<()> {
if bytes.len() > MAX_PENDING_SYNC_ITEM_BYTES {
return Err(BackendError::new(
ClientErrorCategory::Internal,
"sync journal item exceeds safety limit",
));
}
append_u32(payload, bytes.len())?;
payload.extend_from_slice(bytes);
Ok(())
}
fn read_u32(payload: &[u8], offset: &mut usize) -> BackendResult<usize> {
let end = offset.saturating_add(4);
let bytes = payload
.get(*offset..end)
.ok_or_else(|| invalid_pending_payload("journal is truncated"))?;
*offset = end;
Ok(u32::from_le_bytes(
bytes
.try_into()
.map_err(|_| invalid_pending_payload("length is invalid"))?,
) as usize)
}
fn read_bytes<'a>(payload: &'a [u8], offset: &mut usize) -> BackendResult<&'a [u8]> {
let length = read_u32(payload, offset)?;
if length > MAX_PENDING_SYNC_ITEM_BYTES {
return Err(invalid_pending_payload("journal item exceeds safety limit"));
}
let end = offset.saturating_add(length);
let bytes = payload
.get(*offset..end)
.ok_or_else(|| invalid_pending_payload("journal item is truncated"))?;
*offset = end;
Ok(bytes)
}
fn invalid_pending_payload(message: &'static str) -> BackendError {
BackendError::new(
ClientErrorCategory::ProtocolMismatch,
format!("invalid pending sync journal: {message}"),
)
}
#[cfg(test)]
mod tests {
use std::{
collections::VecDeque,
sync::{
Arc,
atomic::{AtomicBool, AtomicUsize, Ordering},
},
};
use super::*;
use crate::InMemoryStore;
#[derive(Clone, Debug)]
struct FakeHost {
state: proto::GetUpdatesStateResult,
responses: Arc<Mutex<VecDeque<BackendResult<proto::GetUpdatesResult>>>>,
requests: Arc<Mutex<Vec<proto::GetUpdatesInput>>>,
applied: Arc<Mutex<Vec<Vec<proto::Update>>>>,
repaired_buckets: Arc<Mutex<Vec<SyncBucketKey>>>,
fail_apply: Arc<AtomicBool>,
fetch_delay_ms: u64,
fetches_in_flight: Arc<AtomicUsize>,
max_fetches_in_flight: Arc<AtomicUsize>,
}
impl FakeHost {
fn new(responses: Vec<proto::GetUpdatesResult>) -> Self {
Self {
state: proto::GetUpdatesStateResult {
date: 100,
updates_found: Some(true),
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
},
responses: Arc::new(Mutex::new(
responses.into_iter().map(Ok).collect::<VecDeque<_>>(),
)),
requests: Arc::new(Mutex::new(Vec::new())),
applied: Arc::new(Mutex::new(Vec::new())),
repaired_buckets: Arc::new(Mutex::new(Vec::new())),
fail_apply: Arc::new(AtomicBool::new(false)),
fetch_delay_ms: 0,
fetches_in_flight: Arc::new(AtomicUsize::new(0)),
max_fetches_in_flight: Arc::new(AtomicUsize::new(0)),
}
}
fn failing_apply(mut self) -> Self {
self.fail_apply = Arc::new(AtomicBool::new(true));
self
}
fn with_fetch_delay(mut self, delay_ms: u64) -> Self {
self.fetch_delay_ms = delay_ms;
self
}
}
impl SyncHost for FakeHost {
fn get_updates_state(
&self,
_date: i64,
) -> BoxFuture<'static, BackendResult<proto::GetUpdatesStateResult>> {
let state = self.state.clone();
Box::pin(async move { Ok(state) })
}
fn get_updates(
&self,
input: proto::GetUpdatesInput,
) -> BoxFuture<'static, BackendResult<proto::GetUpdatesResult>> {
let host = self.clone();
Box::pin(async move {
let in_flight = host.fetches_in_flight.fetch_add(1, Ordering::SeqCst) + 1;
host.max_fetches_in_flight
.fetch_max(in_flight, Ordering::SeqCst);
if host.fetch_delay_ms > 0 {
tokio::time::sleep(std::time::Duration::from_millis(host.fetch_delay_ms)).await;
}
host.requests.lock().await.push(input);
let response = host.responses.lock().await.pop_front().unwrap_or_else(|| {
Err(BackendError::new(
ClientErrorCategory::Internal,
"missing fake getUpdates response",
))
});
host.fetches_in_flight.fetch_sub(1, Ordering::SeqCst);
response
})
}
fn apply_sync_batch(
&self,
updates: Vec<proto::Update>,
_sidecars: Option<proto::UpdateSidecars>,
) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>> {
let host = self.clone();
Box::pin(async move {
if host.fail_apply.load(Ordering::Relaxed) {
return Err(BackendError::new(
ClientErrorCategory::Internal,
"fake apply failure",
));
}
host.applied.lock().await.push(updates.clone());
Ok(updates
.iter()
.filter_map(|update| match update.update.as_ref() {
Some(proto::update::Update::NewMessage(update)) => {
let message = update.message.as_ref()?;
Some(ClientEvent::MessageUpserted {
chat_id: InlineId::new(message.chat_id),
message_id: InlineId::new(message.id),
})
}
_ => None,
})
.collect())
})
}
fn repair_bucket(
&self,
key: SyncBucketKey,
) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>> {
let host = self.clone();
Box::pin(async move {
host.repaired_buckets.lock().await.push(key);
Ok(Vec::new())
})
}
}
#[tokio::test]
async fn hint_only_recovery_pages_to_target_and_persists_cursor() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
store
.save_sync_bucket_state(key, SyncBucketState { seq: 1, date: 10 })
.await
.unwrap();
let host = FakeHost::new(vec![
updates_result(vec![message_update(2, 20, 7, 102)], 2, 20, false),
updates_result(vec![message_update(3, 30, 7, 103)], 3, 30, true),
]);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
let events = sync
.process_realtime(&host, vec![chat_hint(7, 3)])
.await
.unwrap();
assert_eq!(events.len(), 2);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 3, date: 30 }
);
let requests = host.requests.lock().await;
assert_eq!(requests.len(), 2);
assert_eq!(requests[0].start_seq, 1);
assert_eq!(requests[0].seq_end, 3);
assert_eq!(requests[1].start_seq, 2);
assert_eq!(requests[1].seq_end, 3);
}
#[tokio::test]
async fn safe_page_accepts_future_revision_metadata() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
store
.save_sync_bucket_state(key, SyncBucketState { seq: 1, date: 10 })
.await
.unwrap();
let mut response = updates_result(vec![message_update(2, 20, 7, 102)], 2, 20, true);
response.core_sync_schema_revision = crate::CORE_SYNC_SCHEMA_REVISION + 1;
let host = FakeHost::new(vec![response]);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
let events = sync
.process_realtime(&host, vec![chat_hint(7, 2)])
.await
.unwrap();
assert_eq!(events.len(), 1);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 2, date: 20 }
);
}
#[tokio::test]
async fn discovery_accepts_unversioned_state_and_page_metadata() {
let store = Arc::new(InMemoryStore::new());
let mut response = skipped_result(0, 1, 20);
response.core_sync_schema_revision = 0;
let mut host = FakeHost::new(vec![response]);
host.state.core_sync_schema_revision = 0;
let sync = SyncManager::new(store.clone(), SyncConfig::default());
let events = sync.discover(&host).await.unwrap();
assert!(events.is_empty());
assert_eq!(
store.sync_bucket_state(SyncBucketKey::User).await.unwrap(),
SyncBucketState { seq: 1, date: 20 }
);
}
#[tokio::test]
async fn warm_catchup_commits_progress_across_more_than_128_pages() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
store
.save_sync_bucket_state(key, SyncBucketState { seq: 1, date: 10 })
.await
.unwrap();
let responses = (2..=131)
.map(|seq| {
updates_result(
vec![message_update(seq, i64::from(seq), 7, i64::from(seq) + 100)],
i64::from(seq),
i64::from(seq),
seq == 131,
)
})
.collect::<Vec<_>>();
let host = FakeHost::new(responses);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
let events = sync
.process_realtime(&host, vec![chat_hint(7, 131)])
.await
.unwrap();
assert_eq!(events.len(), 130);
assert_eq!(host.requests.lock().await.len(), 130);
assert_eq!(host.applied.lock().await.len(), 130);
assert_eq!(store.sync_bucket_state(key).await.unwrap().seq, 131);
assert!(store.pending_sync_batches().await.unwrap().is_empty());
}
#[tokio::test]
async fn gap_recovery_merges_buffered_realtime_update_before_commit() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
store
.save_sync_bucket_state(key, SyncBucketState { seq: 1, date: 10 })
.await
.unwrap();
let host = FakeHost::new(vec![
updates_result(vec![message_update(2, 20, 7, 102)], 2, 20, false),
updates_result(vec![message_update(3, 30, 7, 103)], 3, 30, true),
]);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
sync.process_realtime(&host, vec![message_update(3, 30, 7, 103)])
.await
.unwrap();
let applied = host.applied.lock().await;
assert_eq!(applied.len(), 2);
assert_eq!(
applied
.iter()
.flat_map(|batch| batch.iter().map(update_seq))
.collect::<Vec<_>>(),
vec![2, 3]
);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 3, date: 30 }
);
}
#[tokio::test]
async fn apply_failure_does_not_advance_bucket_cursor() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
store
.save_sync_bucket_state(key, SyncBucketState { seq: 1, date: 10 })
.await
.unwrap();
let host = FakeHost::new(vec![updates_result(
vec![message_update(2, 20, 7, 101)],
2,
20,
true,
)])
.failing_apply();
let sync = SyncManager::new(store.clone(), SyncConfig::default());
let error = sync
.process_realtime(&host, vec![chat_hint(7, 2)])
.await
.unwrap_err();
assert_eq!(error.category, ClientErrorCategory::Internal);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 1, date: 10 }
);
assert_eq!(store.pending_sync_batches().await.unwrap().len(), 1);
host.fail_apply.store(false, Ordering::Relaxed);
let recovered = sync.process_realtime(&host, Vec::new()).await.unwrap();
assert_eq!(recovered.len(), 1);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 2, date: 20 }
);
assert!(store.pending_sync_batches().await.unwrap().is_empty());
}
#[test]
fn pending_sync_payload_round_trips_updates_and_sidecars() {
let updates = vec![message_update(2, 20, 7, 102)];
let sidecars = proto::UpdateSidecars {
users: vec![proto::User {
id: 42,
..Default::default()
}],
..Default::default()
};
let payload = encode_pending_payload(&updates, Some(&sidecars)).unwrap();
let (decoded_updates, decoded_sidecars) = decode_pending_payload(&payload).unwrap();
assert_eq!(decoded_updates, updates);
assert_eq!(decoded_sidecars, Some(sidecars));
}
#[tokio::test]
async fn incompatible_pending_journal_is_discarded_and_refetched_without_advancing_first() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
store
.save_sync_bucket_state(key, SyncBucketState { seq: 1, date: 10 })
.await
.unwrap();
let invalid = proto::Update {
seq: Some(2),
date: Some(20),
update: None,
};
store
.save_pending_sync_batch(PendingSyncBatch {
key,
committed_state: SyncBucketState { seq: 2, date: 20 },
payload: encode_pending_payload(&[invalid], None).unwrap(),
})
.await
.unwrap();
let host = FakeHost::new(vec![updates_result(
vec![message_update(2, 20, 7, 102)],
2,
20,
true,
)]);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
let events = sync.process_realtime(&host, Vec::new()).await.unwrap();
assert_eq!(events.len(), 1);
assert_eq!(host.requests.lock().await.len(), 1);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 2, date: 20 }
);
assert!(store.pending_sync_batches().await.unwrap().is_empty());
}
#[test]
fn page_envelope_rejects_missing_duplicate_and_out_of_range_sequences() {
let missing = updates_result(vec![message_update(2, 20, 7, 102)], 3, 30, true);
assert!(validate_response_page(1, &missing).is_err());
let duplicate = updates_result(
vec![message_update(2, 20, 7, 102), message_update(2, 20, 7, 102)],
2,
20,
true,
);
assert!(validate_response_page(1, &duplicate).is_err());
let out_of_range = updates_result(vec![message_update(4, 40, 7, 104)], 3, 40, true);
assert!(validate_response_page(1, &out_of_range).is_err());
}
#[test]
fn page_envelope_accepts_explicit_irrelevant_sequence_accounting() {
let mut response = updates_result(vec![message_update(3, 30, 7, 103)], 3, 30, true);
response.skipped_sequences = vec![proto::SyncSkippedSequence {
seq: 2,
reason: proto::sync_skipped_sequence::Reason::IrrelevantToBucket as i32,
}];
assert_eq!(
validate_response_page(1, &response).unwrap(),
PageAccounting::default()
);
}
#[tokio::test]
async fn empty_result_behind_hint_is_retried_without_advancing() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
store
.save_sync_bucket_state(key, SyncBucketState { seq: 5, date: 50 })
.await
.unwrap();
let empty = || proto::GetUpdatesResult {
updates: Vec::new(),
seq: 5,
date: 80,
r#final: Some(true),
result_type: proto::get_updates_result::ResultType::Empty as i32,
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
..Default::default()
};
let host = FakeHost::new(vec![empty(), empty(), empty()]);
let sync = SyncManager::new(
store.clone(),
SyncConfig {
inconsistent_empty_retry_attempts: 2,
inconsistent_empty_retry_delay_ms: 0,
..SyncConfig::default()
},
);
let error = sync
.process_realtime(&host, vec![chat_hint(7, 8)])
.await
.unwrap_err();
assert_eq!(error.category, ClientErrorCategory::ProtocolMismatch);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 5, date: 50 }
);
}
#[tokio::test]
async fn cold_chat_too_long_repairs_snapshot_and_advances_to_hint() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
let host = FakeHost::new(vec![
proto::GetUpdatesResult {
updates: Vec::new(),
seq: 400,
date: 80,
r#final: Some(false),
result_type: proto::get_updates_result::ResultType::TooLong as i32,
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
..Default::default()
},
skipped_result(400, 500, 90),
]);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
sync.process_realtime(&host, vec![chat_hint(7, 500)])
.await
.unwrap();
assert_eq!(host.repaired_buckets.lock().await.as_slice(), &[key]);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 500, date: 90 }
);
assert_eq!(host.requests.lock().await.len(), 2);
}
#[tokio::test]
async fn cold_pointer_commits_lossless_events_without_an_empty_journal() {
let store = Arc::new(InMemoryStore::new());
let key = chat_key(7);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
let event = ClientEvent::MessageDeleted {
chat_id: InlineId::new(7),
message_id: InlineId::new(101),
};
let deliveries = sync
.commit_cold_pointer(key, 500, 80, vec![event.clone()])
.await
.unwrap();
assert_eq!(deliveries.len(), 1);
assert_eq!(deliveries[0].event, event);
assert!(deliveries[0].delivery_id.is_some());
assert!(store.pending_sync_batches().await.unwrap().is_empty());
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 500, date: 80 }
);
assert_eq!(store.pending_client_events().await.unwrap(), deliveries);
}
#[tokio::test]
async fn cold_user_and_space_too_long_require_snapshots_before_advancing() {
for key in [
SyncBucketKey::User,
SyncBucketKey::Space {
space_id: InlineId::new(9),
},
] {
let store = Arc::new(InMemoryStore::new());
let host = FakeHost::new(vec![
proto::GetUpdatesResult {
updates: Vec::new(),
seq: 400,
date: 80,
r#final: Some(false),
result_type: proto::get_updates_result::ResultType::TooLong as i32,
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
..Default::default()
},
skipped_result(400, 500, 90),
]);
let sync = SyncManager::new(store.clone(), SyncConfig::default());
sync.fetch_bucket(
&host,
key,
SyncBucketState::default(),
BTreeMap::new(),
Some(500),
)
.await
.unwrap();
assert_eq!(host.repaired_buckets.lock().await.as_slice(), &[key]);
assert_eq!(
store.sync_bucket_state(key).await.unwrap(),
SyncBucketState { seq: 500, date: 90 }
);
}
}
#[tokio::test]
async fn independent_bucket_fetches_use_configured_concurrency() {
let store = Arc::new(InMemoryStore::new());
let empty = |date| proto::GetUpdatesResult {
updates: Vec::new(),
seq: 1,
date,
r#final: Some(true),
result_type: proto::get_updates_result::ResultType::Empty as i32,
skipped_sequences: vec![proto::SyncSkippedSequence {
seq: 1,
reason: proto::sync_skipped_sequence::Reason::IrrelevantToBucket as i32,
}],
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
..Default::default()
};
let host = FakeHost::new(vec![empty(100), empty(200)]).with_fetch_delay(25);
let sync = SyncManager::new(
store.clone(),
SyncConfig {
max_concurrent_bucket_fetches: 2,
..SyncConfig::default()
},
);
sync.process_realtime(&host, vec![chat_hint(7, 1), chat_hint(8, 1)])
.await
.unwrap();
assert_eq!(host.max_fetches_in_flight.load(Ordering::SeqCst), 2);
assert_eq!(store.sync_bucket_state(chat_key(7)).await.unwrap().seq, 1);
assert_eq!(store.sync_bucket_state(chat_key(8)).await.unwrap().seq, 1);
assert_eq!(store.sync_state().await.unwrap().last_sync_date, 185);
}
fn updates_result(
updates: Vec<proto::Update>,
seq: i64,
date: i64,
final_page: bool,
) -> proto::GetUpdatesResult {
proto::GetUpdatesResult {
updates,
seq,
date,
r#final: Some(final_page),
result_type: proto::get_updates_result::ResultType::Slice as i32,
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
..Default::default()
}
}
fn skipped_result(start_seq: i64, end_seq: i64, date: i64) -> proto::GetUpdatesResult {
proto::GetUpdatesResult {
updates: Vec::new(),
seq: end_seq,
date,
r#final: Some(true),
result_type: proto::get_updates_result::ResultType::Empty as i32,
skipped_sequences: ((start_seq + 1)..=end_seq)
.map(|seq| proto::SyncSkippedSequence {
seq,
reason: proto::sync_skipped_sequence::Reason::IrrelevantToBucket as i32,
})
.collect(),
core_sync_schema_revision: crate::CORE_SYNC_SCHEMA_REVISION,
..Default::default()
}
}
fn chat_hint(chat_id: i64, seq: i32) -> proto::Update {
proto::Update {
seq: None,
date: None,
update: Some(proto::update::Update::ChatHasNewUpdates(
proto::UpdateChatHasNewUpdates {
chat_id,
update_seq: seq,
peer_id: Some(chat_peer(chat_id)),
},
)),
}
}
fn message_update(seq: i32, date: i64, chat_id: i64, message_id: i64) -> proto::Update {
proto::Update {
seq: Some(seq),
date: Some(date),
update: Some(proto::update::Update::NewMessage(proto::UpdateNewMessage {
message: Some(proto::Message {
id: message_id,
chat_id,
peer_id: Some(chat_peer(chat_id)),
..Default::default()
}),
})),
}
}
fn chat_peer(chat_id: i64) -> proto::Peer {
proto::Peer {
r#type: Some(proto::peer::Type::Chat(proto::PeerChat { chat_id })),
}
}
}