Keep the fully-read marker monotonic

src/api/client/read_marker/mod.rs

@@ -1,12 +1,73 @@
 mod read_markers;
 mod receipt;
 
-use ruma::{EventId, MilliSecondsSinceUnixEpoch, RoomId, UserId, events::receipt::ReceiptThread};
+use ruma::{
+	EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, UserId,
+	events::{
+		RoomAccountDataEventType,
+		fully_read::{FullyReadEvent, FullyReadEventContent},
+		receipt::ReceiptThread,
+	},
+};
 use tuwunel_core::{Err, PduCount, Result, err, utils::result::LogErr};
 use tuwunel_service::{Services, rooms::read_receipt::PrivateRead};
 
 pub(crate) use self::{read_markers::set_read_marker_route, receipt::create_receipt_route};
 
+/// Stores the fully-read marker unless it would move the position backwards:
+/// a stale device must not regress the marker another one already advanced.
+/// An unresolvable position on either side accepts the write, as for public
+/// receipts.
+async fn set_fully_read(
+	services: &Services,
+	room_id: &RoomId,
+	user_id: &UserId,
+	event: &EventId,
+) -> Result<()> {
+	let current: Option<OwnedEventId> = services
+		.account_data
+		.get_room::<serde_json::Value>(room_id, user_id, RoomAccountDataEventType::FullyRead)
+		.await
+		.ok()
+		.and_then(|value| {
+			let event_id = value.pointer("/content/event_id")?.as_str()?;
+
+			EventId::parse(event_id).ok()
+		});
+
+	let advances = match current {
+		| Some(current) => {
+			let current = services
+				.timeline
+				.get_pdu_count(&current)
+				.await
+				.ok();
+			let incoming = services.timeline.get_pdu_count(event).await.ok();
+
+			current
+				.zip(incoming)
+				.is_none_or(|(current, incoming)| incoming > current)
+		},
+		| None => true,
+	};
+
+	if !advances {
+		return Ok(());
+	}
+
+	services
+		.account_data
+		.update(
+			Some(room_id),
+			user_id,
+			RoomAccountDataEventType::FullyRead,
+			&serde_json::to_value(FullyReadEvent {
+				content: FullyReadEventContent { event_id: event.to_owned() },
+			})?,
+		)
+		.await
+}
+
 /// Resolves `event` to its timeline position and stores the private read
 /// marker for `thread` there.
 ///

src/api/client/read_marker/read_markers.rs

@@ -4,16 +4,12 @@ use axum::extract::State;
 use ruma::{
 	MilliSecondsSinceUnixEpoch,
 	api::client::read_marker::set_read_marker,
-	events::{
-		RoomAccountDataEventType,
-		fully_read::{FullyReadEvent, FullyReadEventContent},
-		receipt::{Receipt, ReceiptEvent, ReceiptEventContent, ReceiptThread, ReceiptType},
-	},
+	events::receipt::{Receipt, ReceiptEvent, ReceiptEventContent, ReceiptThread, ReceiptType},
 };
 use tuwunel_core::Result;
 use tuwunel_service::presence::Ping;
 
-use super::{reset_and_refresh_badge, set_private_marker};
+use super::{reset_and_refresh_badge, set_fully_read, set_private_marker};
 use crate::{ClientIp, Ruma};
 
 /// # `POST /_matrix/client/r0/rooms/{roomId}/read_markers`
@@ -31,20 +27,7 @@ pub(crate) async fn set_read_marker_route(
 	let sender_user = body.sender_user();
 
 	if let Some(event) = &body.fully_read {
-		let fully_read_event = FullyReadEvent {
-			content: FullyReadEventContent { event_id: event.clone() },
-		};
-
-		services
-			.account_data
-			.update(
-				Some(&body.room_id),
-				sender_user,
-				RoomAccountDataEventType::FullyRead,
-				&serde_json::to_value(fully_read_event)?,
-			)
-			.await
-			.ok();
+		set_fully_read(&services, &body.room_id, sender_user, event).await?;
 	}
 
 	let private_advanced = match &body.private_read_receipt {

src/api/client/read_marker/receipt.rs

@@ -4,16 +4,12 @@ use axum::extract::State;
 use ruma::{
 	MilliSecondsSinceUnixEpoch,
 	api::client::receipt::create_receipt::{self, v3::ReceiptType as CreateReceiptType},
-	events::{
-		RoomAccountDataEventType,
-		fully_read::{FullyReadEvent, FullyReadEventContent},
-		receipt::{Receipt, ReceiptEvent, ReceiptEventContent, ReceiptThread, ReceiptType},
-	},
+	events::receipt::{Receipt, ReceiptEvent, ReceiptEventContent, ReceiptThread, ReceiptType},
 };
 use tuwunel_core::{Err, Result};
 use tuwunel_service::presence::Ping;
 
-use super::{reset_and_refresh_badge, set_private_marker};
+use super::{reset_and_refresh_badge, set_fully_read, set_private_marker};
 use crate::{ClientIp, Ruma};
 
 /// # `POST /_matrix/client/r0/rooms/{roomId}/receipt/{receiptType}/{eventId}`
@@ -72,18 +68,7 @@ pub(crate) async fn create_receipt_route(
 
 	let advanced = match body.receipt_type {
 		| CreateReceiptType::FullyRead => {
-			let fully_read_event = FullyReadEvent {
-				content: FullyReadEventContent { event_id: body.event_id.clone() },
-			};
-			services
-				.account_data
-				.update(
-					Some(&body.room_id),
-					sender_user,
-					RoomAccountDataEventType::FullyRead,
-					&serde_json::to_value(fully_read_event)?,
-				)
-				.await?;
+			set_fully_read(&services, &body.room_id, sender_user, &body.event_id).await?;
 
 			false
 		},