Refactor sliding-sync.

Signed-off-by: Jason Volk <jason@zemos.net>

Cargo.toml

@@ -436,6 +436,7 @@ version = "2.7"
 version = "0.3"
 features = [
 	"ffi",
+	"serde",
 	"std",
 	"union",
 ]

src/service/sync/mod.rs

@@ -5,8 +5,11 @@ use std::{
 	sync::{Arc, Mutex, Mutex as StdMutex},
 };
 
-use ruma::{OwnedDeviceId, OwnedRoomId, OwnedUserId, api::client::sync::sync_events::v5};
+use ruma::{
-use tuwunel_core::Result;
+	OwnedDeviceId, OwnedRoomId, OwnedUserId,
+	api::client::sync::sync_events::v5::{Request, request},
+};
+use tuwunel_core::{Result, implement, smallstr::SmallString};
 use tuwunel_database::Map;
 
 pub struct Service {
@@ -31,17 +34,21 @@ pub struct Data {
 	userid_lastonetimekeyupdate: Arc<Map>,
 }
 
-#[derive(Default)]
+#[derive(Debug, Default)]
 struct SnakeSyncCache {
-	lists: BTreeMap<String, v5::request::List>,
+	lists: BTreeMap<ListId, request::List>,
-	subscriptions: BTreeMap<OwnedRoomId, v5::request::RoomSubscription>,
+	subscriptions: RoomSubscriptions,
-	known_rooms: BTreeMap<String, BTreeMap<OwnedRoomId, u64>>,
+	known_rooms: KnownRooms,
-	extensions: v5::request::Extensions,
+	extensions: request::Extensions,
 }
 
-type DbConnections<K, V> = Mutex<BTreeMap<K, V>>;
+pub type KnownRooms = BTreeMap<ListId, BTreeMap<OwnedRoomId, u64>>;
-type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option<String>);
+pub type RoomSubscriptions = BTreeMap<OwnedRoomId, request::RoomSubscription>;
+pub type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option<ConnId>);
 type SnakeConnectionsVal = Arc<Mutex<SnakeSyncCache>>;
+type DbConnections<K, V> = Mutex<BTreeMap<K, V>>;
+pub type ListId = SmallString<[u8; 16]>;
+pub type ConnId = SmallString<[u8; 16]>;
 
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
@@ -69,185 +76,189 @@ impl crate::Service for Service {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
-impl Service {
+#[implement(Service)]
-	pub fn snake_connection_cached(&self, key: &SnakeConnectionsKey) -> bool {
+pub fn update_snake_sync_request_with_cache(
-		self.snake_connections
+	&self,
-			.lock()
+	snake_key: &SnakeConnectionsKey,
-			.expect("locked")
+	request: &mut Request,
-			.contains_key(key)
+) -> KnownRooms {
-	}
+	let mut cache = self.snake_connections.lock().expect("locked");
+	let cached = Arc::clone(
+		cache
+			.entry(snake_key.clone())
+			.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
+	);
 
-	pub fn forget_snake_sync_connection(&self, key: &SnakeConnectionsKey) {
+	let cached = &mut cached.lock().expect("locked");
-		self.snake_connections
+	drop(cache);
-			.lock()
-			.expect("locked")
-			.remove(key);
-	}
 
-	pub fn update_snake_sync_request_with_cache(
+	//Request::try_from_http_request(req, path_args);
-		&self,
+	for (list_id, list) in &mut request.lists {
-		snake_key: &SnakeConnectionsKey,
+		if let Some(cached_list) = cached.lists.get(list_id.as_str()) {
-		request: &mut v5::Request,
+			list_or_sticky(
-	) -> BTreeMap<String, BTreeMap<OwnedRoomId, u64>> {
+				&mut list.room_details.required_state,
-		let mut cache = self.snake_connections.lock().expect("locked");
+				&cached_list.room_details.required_state,
-		let cached = Arc::clone(
+			);
-			cache
-				.entry(snake_key.clone())
-				.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
-		);
-		let cached = &mut cached.lock().expect("locked");
-		drop(cache);
 
-		//v5::Request::try_from_http_request(req, path_args);
+			//some_or_sticky(&mut list.include_heroes, cached_list.include_heroes);
-		for (list_id, list) in &mut request.lists {
-			if let Some(cached_list) = cached.lists.get(list_id) {
-				list_or_sticky(
-					&mut list.room_details.required_state,
-					&cached_list.room_details.required_state,
-				);
 
-				//some_or_sticky(&mut list.include_heroes, cached_list.include_heroes);
+			match (&mut list.filters, cached_list.filters.clone()) {
-
+				| (Some(filters), Some(cached_filters)) => {
-				match (&mut list.filters, cached_list.filters.clone()) {
+					some_or_sticky(&mut filters.is_invite, cached_filters.is_invite);
-					| (Some(filters), Some(cached_filters)) => {
+					// TODO (morguldir): Find out how a client can unset this, probably need
-						some_or_sticky(&mut filters.is_invite, cached_filters.is_invite);
+					// to change into an option inside ruma
-						// TODO (morguldir): Find out how a client can unset this, probably need
+					list_or_sticky(&mut filters.not_room_types, &cached_filters.not_room_types);
-						// to change into an option inside ruma
+				},
-						list_or_sticky(
+				| (_, Some(cached_filters)) => list.filters = Some(cached_filters),
-							&mut filters.not_room_types,
+				| (Some(list_filters), _) => list.filters = Some(list_filters.clone()),
-							&cached_filters.not_room_types,
+				| (..) => {},
-						);
-					},
-					| (_, Some(cached_filters)) => list.filters = Some(cached_filters),
-					| (Some(list_filters), _) => list.filters = Some(list_filters.clone()),
-					| (..) => {},
-				}
 			}
-			cached.lists.insert(list_id.clone(), list.clone());
 		}
 
 		cached
-			.subscriptions
-			.extend(request.room_subscriptions.clone());
-		request
-			.room_subscriptions
-			.extend(cached.subscriptions.clone());
-
-		request.extensions.e2ee.enabled = request
-			.extensions
-			.e2ee
-			.enabled
-			.or(cached.extensions.e2ee.enabled);
-
-		request.extensions.to_device.enabled = request
-			.extensions
-			.to_device
-			.enabled
-			.or(cached.extensions.to_device.enabled);
-
-		request.extensions.account_data.enabled = request
-			.extensions
-			.account_data
-			.enabled
-			.or(cached.extensions.account_data.enabled);
-		request.extensions.account_data.lists = request
-			.extensions
-			.account_data
 			.lists
-			.clone()
+			.insert(list_id.as_str().into(), list.clone());
-			.or_else(|| cached.extensions.account_data.lists.clone());
-		request.extensions.account_data.rooms = request
-			.extensions
-			.account_data
-			.rooms
-			.clone()
-			.or_else(|| cached.extensions.account_data.rooms.clone());
-
-		some_or_sticky(&mut request.extensions.typing.enabled, cached.extensions.typing.enabled);
-		some_or_sticky(
-			&mut request.extensions.typing.rooms,
-			cached.extensions.typing.rooms.clone(),
-		);
-		some_or_sticky(
-			&mut request.extensions.typing.lists,
-			cached.extensions.typing.lists.clone(),
-		);
-		some_or_sticky(
-			&mut request.extensions.receipts.enabled,
-			cached.extensions.receipts.enabled,
-		);
-		some_or_sticky(
-			&mut request.extensions.receipts.rooms,
-			cached.extensions.receipts.rooms.clone(),
-		);
-		some_or_sticky(
-			&mut request.extensions.receipts.lists,
-			cached.extensions.receipts.lists.clone(),
-		);
-
-		cached.extensions = request.extensions.clone();
-		cached.known_rooms.clone()
 	}
 
-	pub fn update_snake_sync_known_rooms(
+	cached
-		&self,
+		.subscriptions
-		key: &SnakeConnectionsKey,
+		.extend(request.room_subscriptions.clone());
-		list_id: String,
-		new_cached_rooms: BTreeSet<OwnedRoomId>,
-		globalsince: u64,
-	) {
-		assert!(key.2.is_some(), "Some(conn_id) required for this call");
-		let mut cache = self.snake_connections.lock().expect("locked");
-		let cached = Arc::clone(
-			cache
-				.entry(key.clone())
-				.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
-		);
-		let cached = &mut cached.lock().expect("locked");
-		drop(cache);
 
-		for (room_id, lastsince) in cached
+	request
-			.known_rooms
+		.room_subscriptions
-			.entry(list_id.clone())
+		.extend(cached.subscriptions.clone());
-			.or_default()
+
-			.iter_mut()
+	request.extensions.e2ee.enabled = request
-		{
+		.extensions
-			if !new_cached_rooms.contains(room_id) {
+		.e2ee
-				*lastsince = 0;
+		.enabled
-			}
+		.or(cached.extensions.e2ee.enabled);
-		}
+
-		let list = cached.known_rooms.entry(list_id).or_default();
+	request.extensions.to_device.enabled = request
-		for room_id in new_cached_rooms {
+		.extensions
-			list.insert(room_id, globalsince);
+		.to_device
+		.enabled
+		.or(cached.extensions.to_device.enabled);
+
+	request.extensions.account_data.enabled = request
+		.extensions
+		.account_data
+		.enabled
+		.or(cached.extensions.account_data.enabled);
+	request.extensions.account_data.lists = request
+		.extensions
+		.account_data
+		.lists
+		.clone()
+		.or_else(|| cached.extensions.account_data.lists.clone());
+	request.extensions.account_data.rooms = request
+		.extensions
+		.account_data
+		.rooms
+		.clone()
+		.or_else(|| cached.extensions.account_data.rooms.clone());
+
+	{
+		let (request, cached) = (&mut request.extensions.typing, &cached.extensions.typing);
+		some_or_sticky(&mut request.enabled, cached.enabled);
+		some_or_sticky(&mut request.rooms, cached.rooms.clone());
+		some_or_sticky(&mut request.lists, cached.lists.clone());
+	};
+	{
+		let (request, cached) = (&mut request.extensions.receipts, &cached.extensions.receipts);
+		some_or_sticky(&mut request.enabled, cached.enabled);
+		some_or_sticky(&mut request.rooms, cached.rooms.clone());
+		some_or_sticky(&mut request.lists, cached.lists.clone());
+	};
+
+	cached.extensions = request.extensions.clone();
+	cached.known_rooms.clone()
+}
+
+#[implement(Service)]
+pub fn update_snake_sync_known_rooms(
+	&self,
+	key: &SnakeConnectionsKey,
+	list_id: ListId,
+	new_cached_rooms: BTreeSet<OwnedRoomId>,
+	globalsince: u64,
+) {
+	assert!(key.2.is_some(), "Some(conn_id) required for this call");
+
+	let mut cache = self.snake_connections.lock().expect("locked");
+	let cached = Arc::clone(
+		cache
+			.entry(key.clone())
+			.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
+	);
+
+	let cached = &mut cached.lock().expect("locked");
+	drop(cache);
+
+	for (room_id, lastsince) in cached
+		.known_rooms
+		.entry(list_id.clone())
+		.or_default()
+		.iter_mut()
+	{
+		if !new_cached_rooms.contains(room_id) {
+			*lastsince = 0;
 		}
 	}
 
-	pub fn update_snake_sync_subscriptions(
+	let list = cached.known_rooms.entry(list_id).or_default();
-		&self,
+	for room_id in new_cached_rooms {
-		key: &SnakeConnectionsKey,
+		list.insert(room_id, globalsince);
-		subscriptions: BTreeMap<OwnedRoomId, v5::request::RoomSubscription>,
-	) {
-		let mut cache = self.snake_connections.lock().expect("locked");
-		let cached = Arc::clone(
-			cache
-				.entry(key.clone())
-				.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
-		);
-		let cached = &mut cached.lock().expect("locked");
-		drop(cache);
-
-		cached.subscriptions = subscriptions;
 	}
 }
 
+#[implement(Service)]
+pub fn update_snake_sync_subscriptions(
+	&self,
+	key: &SnakeConnectionsKey,
+	subscriptions: RoomSubscriptions,
+) {
+	let mut cache = self.snake_connections.lock().expect("locked");
+	let cached = Arc::clone(
+		cache
+			.entry(key.clone())
+			.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
+	);
+
+	let cached = &mut cached.lock().expect("locked");
+	drop(cache);
+
+	cached.subscriptions = subscriptions;
+}
+
+#[implement(Service)]
+pub fn forget_snake_sync_connection(&self, key: &SnakeConnectionsKey) {
+	self.snake_connections
+		.lock()
+		.expect("locked")
+		.remove(key);
+}
+
+#[implement(Service)]
+pub fn snake_connection_cached(&self, key: &SnakeConnectionsKey) -> bool {
+	self.snake_connections
+		.lock()
+		.expect("locked")
+		.contains_key(key)
+}
+
 #[inline]
-pub fn into_snake_key<U, D, C>(user_id: U, device_id: D, conn_id: C) -> SnakeConnectionsKey
+pub fn into_snake_key<U, D, C>(
+	user_id: U,
+	device_id: D,
+	conn_id: Option<C>,
+) -> SnakeConnectionsKey
 where
 	U: Into<OwnedUserId>,
 	D: Into<OwnedDeviceId>,
-	C: Into<Option<String>>,
+	C: Into<ConnId>,
 {
-	(user_id.into(), device_id.into(), conn_id.into())
+	(user_id.into(), device_id.into(), conn_id.map(Into::into))
 }
 
 /// load params from cache if body doesn't contain it, as long as it's allowed