tuwunel_api/client/sync/

v5.rs

mod extensions;
mod filter;
mod range;
mod rooms;
mod selector;

use std::{
	collections::{BTreeMap, BTreeSet},
	fmt::Debug,
	sync::Arc,
	time::Duration,
};

use axum::extract::{Extension, State};
use futures::{FutureExt, TryFutureExt, future::join};
use ruma::{
	DeviceId, OwnedRoomId, UserId,
	api::client::sync::sync_events::v5::{ListId, Request, Response, response},
	events::room::member::MembershipState,
};
use tokio::{
	sync::Notify,
	time::{Instant, timeout_at},
};
use tuwunel_core::{
	Err, Result, debug,
	debug::INFO_SPAN_LEVEL,
	debug_warn,
	error::inspect_log,
	smallvec::SmallVec,
	trace,
	utils::{TryFutureExtExt, result::FlatOk},
};
use tuwunel_service::{
	Services,
	presence::Ping,
	sync::{Connection, into_connection_key},
};

use self::{
	extensions::{apply_profiles, apply_ranges, handle as handle_extensions},
	range::collect as collect_ranges,
};
use super::share_encrypted_room;
use crate::{ClientIp, Ruma};

#[derive(Copy, Clone)]
struct SyncInfo<'a> {
	services: &'a Services,
	sender_user: &'a UserId,
	sender_device: Option<&'a DeviceId>,
	previous_connection_pos: Option<u64>,

	/// The rooms the user calls direct, read once for the whole pass.
	///
	/// Both the list filters and every room payload answer `is_dm` from this.
	direct_rooms: &'a BTreeSet<OwnedRoomId>,
}

#[derive(Clone, Debug)]
struct WindowRoom {
	room_id: OwnedRoomId,
	membership: Option<MembershipState>,
	lists: ListIds,
	event_count: u64,
	payload_count: u64,
}

impl WindowRoom {
	#[inline]
	fn payload_is_fresh(&self, roomsince: u64) -> bool {
		roomsince == 0 || self.payload_count > roomsince
	}
}

type Window = BTreeMap<OwnedRoomId, WindowRoom>;
type ResponseLists = BTreeMap<ListId, response::List>;
type ListIds = SmallVec<[ListId; 1]>;

/// `POST /_matrix/client/unstable/org.matrix.simplified_msc3575/sync`
/// ([MSC4186])
///
/// A simplified version of sliding sync ([MSC3575]).
///
/// Get all new events in a sliding window of rooms since the last sync or a
/// given point in time.
///
/// [MSC3575]: https://github.com/matrix-org/matrix-spec-proposals/pull/3575
/// [MSC4186]: https://github.com/matrix-org/matrix-spec-proposals/pull/4186
#[tracing::instrument(
	name = "sync",
	level = INFO_SPAN_LEVEL,
	skip_all,
	fields(
		user_id = %body.sender_user().localpart(),
		device_id = %body.sender_device.as_deref().map_or("<no device>", |x| x.as_str()),
		conn_id = ?body.body.conn_id.clone().unwrap_or_default(),
		since = ?body.body.pos.clone().unwrap_or_default(),
	)
)]
pub(crate) async fn sync_events_v5_route(
	Extension(interrupted): Extension<Arc<Notify>>,
	ClientIp(client): ClientIp,
	State(ref services): State<crate::State>,
	body: Ruma<Request>,
) -> Result<Response> {
	let sender_user = body.sender_user();
	let sender_device = body.sender_device.as_deref();
	let request = &body.body;
	let since = request
		.pos
		.as_ref()
		.and_then(|string| string.parse().ok())
		.unwrap_or(0);

	let timeout = request
		.timeout
		.as_ref()
		.map(Duration::as_millis)
		.map(TryInto::try_into)
		.flat_ok()
		.map(|timeout: u64| timeout.min(services.config.client_sync_timeout_max))
		.unwrap_or(0);

	let conn_key = into_connection_key(sender_user, sender_device, request.conn_id.as_deref());
	let conn_val = services
		.sync
		.load_or_init_connection(&conn_key)
		.await;

	let conn = conn_val.lock();
	let ping = Ping {
		device_id: sender_device,
		client_ip: Some(client),
		new_state: Some(&request.set_presence),
		appservice: body.appservice_info.as_ref(),
	};

	let ping_presence = services
		.presence
		.maybe_ping_presence(sender_user, ping)
		.inspect_err(inspect_log)
		.ok();

	let (mut conn, _) = join(conn, ping_presence).await;

	if since != 0 && conn.next_batch == 0 {
		return Err!(Request(UnknownPos(warn!("Connection lost; restarting sync stream."))));
	}

	if since == 0 {
		*conn = Connection::default();
		conn.store(&services.sync, &conn_key);
		debug_warn!(?conn_key, "Client cleared cache and reloaded.");
	}

	let advancing = since == conn.next_batch;
	let retarding = since != 0 && since <= conn.globalsince;
	if !advancing && !retarding {
		return Err!(Request(UnknownPos(warn!(
			"Requesting unknown or invalid stream position."
		))));
	}

	debug_assert!(
		advancing || retarding,
		"Request should either be advancing or replaying the since token."
	);

	// Update parameters regardless of replay or advance
	conn.next_batch = services.globals.wait_pending().await?;
	conn.globalsince = since.min(conn.next_batch);
	let config_changed = conn.update_cache(request);
	let caught_up = conn.globalsince == conn.next_batch;

	// A whole profile owed to a caught-up connection needs a pass like a new list.
	let needs_pass = config_changed || conn.own_profile_owed();

	if config_change_needs_position(needs_pass, advancing, caught_up, since) {
		// The permit publishes the reserved position when it retires on drop.
		drop(services.globals.next_count());
	}

	conn.update_rooms_prologue(retarding.then_some(since));

	let mut response = Response {
		txn_id: request.txn_id.clone(),
		lists: Default::default(),
		pos: Default::default(),
		rooms: Default::default(),
		extensions: Default::default(),
	};

	let stop_at = Instant::now()
		.checked_add(Duration::from_millis(timeout))
		.expect("configuration must limit maximum timeout");

	loop {
		debug_assert!(
			conn.globalsince <= conn.next_batch,
			"since should not be greater than next_batch."
		);

		let window;
		let watchers = services
			.sync
			.watch(sender_user, sender_device, services.state_cache.rooms_joined(sender_user))
			.await;

		conn.next_batch = services.globals.wait_pending().await?;
		let direct_rooms = services
			.account_data
			.direct_rooms(sender_user)
			.await;

		let sync_info = SyncInfo {
			services,
			sender_user,
			sender_device,
			previous_connection_pos: since.ne(&0).then_some(since),
			direct_rooms: &direct_rooms,
		};

		(window, response.lists) = selector::selector(&mut conn, sync_info)
			.boxed()
			.await;

		if conn.globalsince < conn.next_batch {
			let ranges = collect_ranges(sync_info, &conn, &window);
			let extensions = handle_extensions(sync_info, &conn, &window);
			let (mut ranges, extensions) = join(ranges, extensions).boxed().await;

			let mut extensions = extensions?;

			apply_profiles(sync_info, &conn, &window, &ranges, &mut extensions).await?;
			apply_ranges(&conn, &window, &mut ranges, &mut extensions);
			conn.update_rooms_epilogue(ranges.room_updates());
			conn.update_profiles_epilogue();
			response.rooms = ranges.into_payloads();
			response.extensions = extensions.into_response(&response.rooms);

			if !is_empty_response(&response) {
				response.pos = conn.next_batch.to_string().into();
				trace!(conn.globalsince, conn.next_batch, "response {response:?}");
				conn.store(&services.sync, &conn_key);
				return Ok(response);
			}
		}

		let waiter = async || {
			tokio::select! {
				() = interrupted.notified() => true,
				watch = timeout_at(stop_at, watchers) => watch.is_err(),
			}
		};

		if timeout == 0 || services.server.is_stopping() || waiter().boxed().await {
			response.pos = conn.next_batch.to_string().into();
			trace!(conn.globalsince, conn.next_batch, "empty response {response:?}");
			conn.store(&services.sync, &conn_key);
			return Ok(response);
		}

		debug!(
			?timeout,
			last_since = conn.globalsince,
			last_batch = conn.next_batch,
			pend_count = ?services.globals.pending_count(),
			"notified by watcher"
		);

		conn.globalsince = conn.next_batch;
	}
}

fn config_change_needs_position(
	config_changed: bool,
	advancing: bool,
	caught_up: bool,
	since: u64,
) -> bool {
	config_changed && advancing && caught_up && since != 0
}

fn is_empty_response(response: &Response) -> bool {
	response.extensions.is_empty() && response.rooms.is_empty()
}

#[cfg(test)]
mod tests {
	use super::config_change_needs_position;

	#[test]
	fn caught_up_advancing_config_change_needs_position() {
		assert!(config_change_needs_position(true, true, true, 1));
		assert!(!config_change_needs_position(false, true, true, 1));
		assert!(!config_change_needs_position(true, false, true, 1));
		assert!(!config_change_needs_position(true, true, false, 1));
		assert!(!config_change_needs_position(true, true, true, 0));
	}
}