feat(sync): Add MSC4354 sticky events

Cargo.toml

@@ -377,6 +377,7 @@ features = [
     "unstable-msc4293",
     "unstable-msc4406",
     "unstable-msc4439",
+    "unstable-msc4354",
     "unstable-msc4466",
     "unstable-extensible-events",
 ]

changelog.d/+cf65f70c.feature.md

@@ -0,0 +1 @@
+Added MSC4354 sticky events, behind the `allow_sticky_events` config option. Contributed by @eleboucher.

conduwuit-example.toml

@@ -1306,6 +1306,12 @@
 #
 #allow_incoming_typing = true
 
+# Allow the use of unstable persistent "sticky events" ([MSC4354])
+#
+# MSC4354: https://c10y.cc/MSC4354
+#
+#allow_sticky_events = false
+
 # Maximum time federation user can indicate typing.
 #
 #typing_federation_timeout_s = 30

src/api/client/room/create.rs

@@ -541,6 +541,7 @@ pub(crate) async fn create_room_route(
 				&content,
 				&state_key,
 				None,
+				None,
 			)
 			.boxed()
 			.await

src/api/client/send.rs

@@ -1,7 +1,11 @@
 use std::collections::BTreeMap;
 
 use axum::extract::State;
-use conduwuit::{Err, Result, err, matrix::pdu::PartialPdu, utils};
+use conduwuit::{
+	Err, Result, err,
+	matrix::pdu::{PartialPdu, sticky},
+	utils,
+};
 use ruma::{api::client::message::send_message_event, events::MessageLikeEventType};
 use serde_json::from_str;
 
@@ -81,6 +85,12 @@ pub(crate) async fn send_message_event_route(
 			PartialPdu {
 				event_type: body.event_type.clone().into(),
 				content,
+				sticky: services
+					.config
+					.allow_sticky_events
+					.then_some(body.sticky_duration_ms)
+					.flatten()
+					.map(sticky::object),
 				unsigned: Some(unsigned),
 				timestamp: if body.identity.is_appservice() {
 					body.timestamp

src/api/client/state.rs

@@ -49,6 +49,7 @@ pub(crate) async fn send_state_event_for_key_route(
 			&body.event_type,
 			&body.body.body,
 			&body.state_key,
+			body.sticky_duration_ms,
 			if body.identity.is_appservice() {
 				body.timestamp
 			} else {

src/api/client/sync/v3/joined.rs

@@ -4,7 +4,7 @@ use conduwuit::{
 	Result, at, debug_warn, err, extract_variant,
 	matrix::{
 		Event,
-		pdu::{PduCount, PduEvent},
+		pdu::{PduCount, PduEvent, sticky},
 	},
 	trace,
 	utils::{
@@ -20,12 +20,12 @@ use futures::{
 	future::{OptionFuture, join, join3, join4, try_join, try_join3},
 };
 use ruma::{
-	OwnedRoomId, OwnedUserId, RoomId, UserId,
+	MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedUserId, RoomId, UserId,
 	api::client::sync::sync_events::{
 		UnreadNotificationsCount,
 		v3::{
 			Ephemeral, JoinedRoom, RoomAccountData, RoomSummary, State as RoomState, StateEvents,
-			Timeline,
+			Sticky, Timeline,
 		},
 	},
 	assign,
@@ -42,7 +42,7 @@ use tokio::pin;
 
 use super::{load_timeline, share_encrypted_room};
 use crate::client::{
-	TimelinePdus, ignored_filter,
+	TimelinePdus, ignored_filter, is_ignored_pdu,
 	sync::v3::{
 		DEFAULT_TIMELINE_LIMIT, DeviceListUpdates, SyncContext, prepare_lazily_loaded_members,
 		state::{build_state_incremental, build_state_initial},
@@ -76,6 +76,7 @@ pub(super) async fn load_joined_room(
 		StateAndTimeline {
 			state_events,
 			timeline,
+			sticky,
 			summary,
 			notification_counts,
 			device_list_updates,
@@ -111,6 +112,7 @@ pub(super) async fn load_joined_room(
 			RoomState::Before(state_events)
 		},
 		ephemeral,
+		sticky,
 		unread_thread_notifications: BTreeMap::new(),
 	});
 
@@ -251,11 +253,82 @@ async fn build_ephemeral(
 struct StateAndTimeline {
 	state_events: Vec<PduEvent>,
 	timeline: Timeline,
+	sticky: Sticky,
 	summary: Option<RoomSummary>,
 	notification_counts: Option<UnreadNotificationsCount>,
 	device_list_updates: DeviceListUpdates,
 }
 
+async fn build_sticky_events(
+	services: &Services,
+	sync_context: SyncContext<'_>,
+	room_id: &RoomId,
+	timeline: &TimelinePdus,
+	joined_since_last_sync: bool,
+) -> Sticky {
+	if !services.config.allow_sticky_events {
+		return Sticky::new();
+	}
+
+	let SyncContext {
+		syncing_user,
+		last_sync_end_count,
+		current_count,
+		..
+	} = sync_context;
+
+	let timeline_event_ids: HashSet<_> =
+		timeline.pdus.iter().map(|(_, pdu)| &pdu.event_id).collect();
+	let now = u64::from(MilliSecondsSinceUnixEpoch::now().get());
+
+	// on join the client needs every unexpired sticky event, not just the ones
+	// since its last sync
+	let since = (!joined_since_last_sync)
+		.then_some(last_sync_end_count)
+		.flatten()
+		.map(PduCount::Normal);
+
+	// nothing older than this can still be sticky, so the walk back never goes far
+	let oldest_sticky_ts = now.saturating_sub(sticky::MAX_DURATION_MS);
+
+	let mut events = Vec::new();
+	let pdus = services
+		.rooms
+		.timeline
+		.pdus_rev(room_id, Some(PduCount::Normal(current_count).saturating_add(1)))
+		.ignore_err();
+
+	pin!(pdus);
+	while let Some((count, mut pdu)) = pdus.next().await {
+		if since.is_some_and(|since| count <= since)
+			|| u64::from(pdu.origin_server_ts) < oldest_sticky_ts
+		{
+			break;
+		}
+
+		let is_sticky = pdu
+			.sticky
+			.as_deref()
+			.is_some_and(|sticky| sticky::is_sticky(pdu.origin_server_ts, sticky, now));
+
+		if !is_sticky
+			|| timeline_event_ids.contains(&pdu.event_id)
+			|| is_ignored_pdu(services, &pdu, syncing_user)
+				.await
+				.unwrap_or(true)
+		{
+			continue;
+		}
+
+		pdu.set_unsigned(Some(syncing_user));
+		events.push(Event::into_format(pdu));
+	}
+
+	events.reverse();
+
+	assign!(Sticky::new(), { events })
+}
+
 /// Compute changes to the room's state and timeline.
 #[tracing::instrument(level = "debug", skip_all)]
 async fn build_state_and_timeline(
@@ -274,6 +347,9 @@ async fn build_state_and_timeline(
 		check_joined_since_last_sync(services, shortstatehashes, sync_context),
 	)
 	.await?;
+	let sticky =
+		build_sticky_events(services, sync_context, room_id, &timeline, joined_since_last_sync)
+			.await;
 
 	let state_events = build_state_events(
 		services,
@@ -342,6 +418,7 @@ async fn build_state_and_timeline(
 			prev_batch: prev_batch.as_ref().map(ToString::to_string),
 			events: filtered_timeline,
 		}),
+		sticky,
 		summary,
 		notification_counts,
 		device_list_updates,

src/api/client/sync/v3/left.rs

@@ -328,6 +328,7 @@ fn create_dummy_leave_event(
 			.expect("Timestamp is valid js_int value"),
 		kind: TimelineEventType::RoomMember,
 		content: RawValue::from_string(r#"{"membership": "leave"}"#.to_owned()).unwrap(),
+		sticky: None,
 		state_key: Some(syncing_user.as_str().into()),
 		unsigned: None,
 		// The following keys are dropped on conversion

src/api/client/unversioned.rs

@@ -21,12 +21,19 @@ use crate::Ruma;
 /// Note: Unstable features are used while developing new features. Clients
 /// should avoid using unstable features in their stable releases
 pub(crate) async fn get_supported_versions_route(
+	State(services): State<crate::State>,
 	_body: Ruma<get_supported_versions::Request>,
 ) -> Result<get_supported_versions::Response> {
-	Ok(assign!(
-		get_supported_versions::Response::new(versions()),
-		{ unstable_features: unstable_features() }
-	))
+	let mut unstable_features = unstable_features();
+
+	if services.config.allow_sticky_events {
+		// sticky events (https://github.com/matrix-org/matrix-spec-proposals/pull/4354)
+		unstable_features.insert("org.matrix.msc4354".to_owned(), true);
+	}
+
+	Ok(assign!(get_supported_versions::Response::new(versions()), {
+		unstable_features
+	}))
 }
 
 /// # `GET /_continuwuity/server_version`

src/core/config/mod.rs

@@ -1526,6 +1526,14 @@ pub struct Config {
 	#[serde(default = "true_fn")]
 	pub allow_incoming_typing: bool,
 
+	/// Allow the use of unstable persistent "sticky events" ([MSC4354])
+	///
+	/// MSC4354: https://c10y.cc/MSC4354
+	///
+	/// default: false
+	#[serde(default)]
+	pub allow_sticky_events: bool,
+
 	/// Maximum time federation user can indicate typing.
 	///
 	/// default: 30

src/core/matrix/event.rs

@@ -181,6 +181,10 @@ pub trait Event: Clone + Debug {
 	/// The state key for this event.
 	fn state_key(&self) -> Option<&str>;
 
+	/// The MSC4354 sticky object, verbatim, if the event has one.
+	#[inline]
+	fn sticky(&self) -> Option<&RawJsonValue> { None }
+
 	/// The event type.
 	fn kind(&self) -> &TimelineEventType;
 

src/core/matrix/event/format.rs

@@ -9,6 +9,7 @@ use ruma::{
 use serde_json::json;
 
 use super::{Event, redact};
+use crate::matrix::pdu::sticky;
 
 pub struct Owned<E: Event>(pub(super) E);
 
@@ -36,6 +37,9 @@ impl<'a, E: Event> From<Ref<'a, E>> for Raw<AnySyncTimelineEvent> {
 		if let Some(state_key) = event.state_key() {
 			json["state_key"] = json!(state_key);
 		}
+		if let Some(sticky_object) = event.sticky() {
+			json[sticky::PDU_KEY] = json!(sticky_object);
+		}
 		if let Some(unsigned) = event.unsigned() {
 			json["unsigned"] = json!(unsigned);
 		}
@@ -67,6 +71,9 @@ impl<'a, E: Event> From<Ref<'a, E>> for Raw<AnyTimelineEvent> {
 		if let Some(state_key) = event.state_key() {
 			json["state_key"] = json!(state_key);
 		}
+		if let Some(sticky_object) = event.sticky() {
+			json[sticky::PDU_KEY] = json!(sticky_object);
+		}
 		if let Some(unsigned) = event.unsigned() {
 			json["unsigned"] = json!(unsigned);
 		}
@@ -98,6 +105,9 @@ impl<'a, E: Event> From<Ref<'a, E>> for Raw<AnyMessageLikeEvent> {
 		if let Some(state_key) = event.state_key() {
 			json["state_key"] = json!(state_key);
 		}
+		if let Some(sticky_object) = event.sticky() {
+			json[sticky::PDU_KEY] = json!(sticky_object);
+		}
 		if let Some(unsigned) = event.unsigned() {
 			json["unsigned"] = json!(unsigned);
 		}

src/core/matrix/pdu.rs

@@ -3,6 +3,7 @@ mod id;
 mod partial;
 mod raw_id;
 mod redact;
+pub mod sticky;
 #[cfg(test)]
 mod tests;
 mod unsigned;
@@ -46,6 +47,11 @@ pub struct Pdu {
 
 	pub content: Box<RawJsonValue>,
 
+	/// MSC4354 sticky object, kept verbatim because it is signed. Interpret it
+	/// with [`sticky`], never by rewriting it.
+	#[serde(rename = "msc4354_sticky", default, skip_serializing_if = "Option::is_none")]
+	pub sticky: Option<Box<RawJsonValue>>,
+
 	#[serde(skip_serializing_if = "Option::is_none")]
 	pub state_key: Option<StateKey>,
 
@@ -137,6 +143,9 @@ impl Event for Pdu {
 	#[inline]
 	fn state_key(&self) -> Option<&str> { self.state_key.as_deref() }
 
+	#[inline]
+	fn sticky(&self) -> Option<&RawJsonValue> { self.sticky.as_deref() }
+
 	#[inline]
 	fn kind(&self) -> &TimelineEventType { &self.kind }
 
@@ -208,6 +217,9 @@ impl Event for &Pdu {
 	#[inline]
 	fn state_key(&self) -> Option<&str> { self.state_key.as_deref() }
 
+	#[inline]
+	fn sticky(&self) -> Option<&RawJsonValue> { self.sticky.as_deref() }
+
 	#[inline]
 	fn kind(&self) -> &TimelineEventType { &self.kind }
 

src/core/matrix/pdu/partial.rs

@@ -20,6 +20,8 @@ pub struct PartialPdu {
 
 	pub content: Box<RawJsonValue>,
 
+	pub sticky: Option<Box<RawJsonValue>>,
+
 	pub unsigned: Option<Unsigned>,
 
 	pub state_key: Option<StateKey>,
@@ -66,6 +68,7 @@ impl Default for PartialPdu {
 		Self {
 			event_type: "m.room.message".into(),
 			content: to_raw_value(&json!({})).unwrap(),
+			sticky: None,
 			unsigned: None,
 			state_key: None,
 			redacts: None,

src/core/matrix/pdu/redact.rs

@@ -11,6 +11,9 @@ impl super::Pdu {
 
 		self.unsigned = None;
 
+		// a redacted sticky event is just a normal event (MSC4354)
+		self.sticky = None;
+
 		let mut content = serde_json::from_str(self.content.get()).map_err(|e| {
 			err!(Request(BadJson("Failed to deserialize content into type: {e}")))
 		})?;

src/core/matrix/pdu/sticky.rs

@@ -0,0 +1,155 @@
+//! Sticky event helpers, as defined by [MSC4354].
+//!
+//! [MSC4354]: https://github.com/matrix-org/matrix-spec-proposals/pull/4354
+
+use ruma::{UInt, events::sticky::StickyDurationMs};
+use serde::Deserialize;
+use serde_json::{
+	json,
+	value::{RawValue as RawJsonValue, to_raw_value},
+};
+
+/// Name of the sticky object at the top level of a PDU. Unstable prefix.
+pub const PDU_KEY: &str = "msc4354_sticky";
+
+/// Name of the expiry hint injected into `unsigned`. Unstable prefix.
+pub const TTL_UNSIGNED_KEY: &str = "msc4354_sticky_duration_ttl_ms";
+
+/// The longest an event may remain sticky (one hour).
+#[allow(clippy::as_conversions)]
+pub const MAX_DURATION_MS: u64 = StickyDurationMs::MAX as u64;
+
+#[derive(Deserialize)]
+struct Object {
+	duration_ms: u64,
+}
+
+/// Build the sticky object for an event we are creating locally.
+#[must_use]
+pub fn object(duration_ms: StickyDurationMs) -> Box<RawJsonValue> {
+	to_raw_value(&json!({ "duration_ms": duration_ms }))
+		.expect("a StickyDurationMs always produces a valid sticky object")
+}
+
+/// The sticky duration of `sticky`, clamped to [`MAX_DURATION_MS`], or `None`
+/// if this is not a valid sticky object.
+///
+/// A malformed object leaves the event unsticky rather than invalidating it, so
+/// a peer more lenient than us cannot split the room DAG.
+#[must_use]
+pub fn duration_ms(sticky: &RawJsonValue) -> Option<u64> {
+	serde_json::from_str::<Object>(sticky.get())
+		.ok()
+		.map(|sticky| sticky.duration_ms.min(MAX_DURATION_MS))
+}
+
+/// The time at which an event stops being sticky, in milliseconds since the
+/// unix epoch, or `None` if the event was never sticky.
+///
+/// MSC4354 starts the clock at `min(received_ts, origin_server_ts)` so a server
+/// cannot extend stickiness by dating its events forward. We keep no receive
+/// time per PDU, so instead we deny stickiness to events dated further ahead
+/// than any event could still be sticky for.
+#[must_use]
+pub fn expires_at(origin_server_ts: UInt, sticky: &RawJsonValue, now: u64) -> Option<u64> {
+	let origin_server_ts = u64::from(origin_server_ts);
+	if origin_server_ts.saturating_sub(now) > MAX_DURATION_MS {
+		return None;
+	}
+
+	let start = origin_server_ts.min(now);
+	duration_ms(sticky).map(|duration_ms| start.saturating_add(duration_ms))
+}
+
+#[must_use]
+pub fn is_sticky(origin_server_ts: UInt, sticky: &RawJsonValue, now: u64) -> bool {
+	expires_at(origin_server_ts, sticky, now).is_some_and(|expires_at| expires_at > now)
+}
+
+#[cfg(test)]
+mod tests {
+	use ruma::{UInt, events::sticky::StickyDurationMs};
+	use serde_json::value::{RawValue as RawJsonValue, to_raw_value};
+
+	use super::{MAX_DURATION_MS, duration_ms, expires_at, is_sticky, object};
+
+	fn raw(json: &serde_json::Value) -> Box<RawJsonValue> { to_raw_value(json).unwrap() }
+
+	fn ts(millis: u64) -> UInt { UInt::try_from(millis).unwrap() }
+
+	#[test]
+	fn duration_of_valid_object() {
+		let sticky = raw(&serde_json::json!({ "duration_ms": 300_000 }));
+		assert_eq!(duration_ms(&sticky), Some(300_000));
+	}
+
+	#[test]
+	fn duration_is_clamped_to_one_hour() {
+		let sticky = raw(&serde_json::json!({ "duration_ms": 9_999_999_999_u64 }));
+		assert_eq!(duration_ms(&sticky), Some(MAX_DURATION_MS));
+	}
+
+	#[test]
+	fn malformed_objects_are_not_sticky() {
+		for malformed in [
+			serde_json::json!({}),
+			serde_json::json!({ "duration_ms": -1 }),
+			serde_json::json!({ "duration_ms": "300000" }),
+			serde_json::json!({ "duration_ms": 3000.5 }),
+			serde_json::json!("nonsense"),
+			serde_json::json!(null),
+		] {
+			let sticky = raw(&malformed);
+			assert_eq!(duration_ms(&sticky), None, "{malformed} should not be sticky");
+		}
+	}
+
+	#[test]
+	fn object_round_trips_through_duration_ms() {
+		let sticky = object(StickyDurationMs::new_clamped(600_000_u32));
+		assert_eq!(duration_ms(&sticky), Some(600_000));
+	}
+
+	#[test]
+	fn expiry_is_origin_server_ts_plus_duration() {
+		let sticky = raw(&serde_json::json!({ "duration_ms": 300_000 }));
+		assert_eq!(expires_at(ts(1_000_000), &sticky, 1_100_000), Some(1_300_000));
+	}
+
+	#[test]
+	fn expiry_uses_the_clamped_duration() {
+		let sticky = raw(&serde_json::json!({ "duration_ms": 9_999_999_999_u64 }));
+		assert_eq!(
+			expires_at(ts(1_000_000), &sticky, 1_000_000),
+			Some(1_000_000 + MAX_DURATION_MS)
+		);
+	}
+
+	#[test]
+	fn timestamps_far_in_the_future_are_not_sticky() {
+		let sticky = raw(&serde_json::json!({ "duration_ms": 300_000 }));
+		let now = 1_000_000;
+
+		// a little ahead of us is tolerated, and starts the clock at `now`
+		let slightly_ahead = ts(now + MAX_DURATION_MS);
+		assert_eq!(expires_at(slightly_ahead, &sticky, now), Some(now + 300_000));
+
+		// beyond the point any honest event could still be sticky, it is not
+		let far_ahead = ts(now + MAX_DURATION_MS + 1);
+		assert_eq!(expires_at(far_ahead, &sticky, now), None);
+	}
+
+	#[test]
+	fn stickiness_ends_at_the_expiry() {
+		let sticky = raw(&serde_json::json!({ "duration_ms": 300_000 }));
+		assert!(is_sticky(ts(1_000_000), &sticky, 1_299_999));
+		assert!(!is_sticky(ts(1_000_000), &sticky, 1_300_000));
+		assert!(!is_sticky(ts(1_000_000), &sticky, 1_300_001));
+	}
+
+	#[test]
+	fn zero_duration_is_never_sticky() {
+		let sticky = raw(&serde_json::json!({ "duration_ms": 0 }));
+		assert!(!is_sticky(ts(1_000_000), &sticky, 1_000_000));
+	}
+}

src/core/matrix/pdu/tests.rs

@@ -15,3 +15,96 @@ fn normal_parse() {
 
 	assert!(!backfilled, "backfilled variant");
 }
+
+fn sticky_pdu(sticky: &serde_json::Value) -> super::Pdu {
+	serde_json::from_value(serde_json::json!({
+		"event_id": "$test:example.com",
+		"room_id": "!test:example.com",
+		"sender": "@test:example.com",
+		"origin_server_ts": 1_000_000,
+		"type": "m.room.message",
+		"content": { "msgtype": "m.text", "body": "test" },
+		"msc4354_sticky": sticky,
+		"prev_events": [],
+		"depth": 1,
+		"auth_events": [],
+		"hashes": { "sha256": "test_hash" },
+	}))
+	.expect("PDU is valid")
+}
+
+#[test]
+fn sticky_object_round_trips_verbatim() {
+	// the object is signed, so serialization must not touch it
+	let pdu = sticky_pdu(&serde_json::json!({ "duration_ms": 300_000, "unknown": "key" }));
+	let value = serde_json::to_value(&pdu).expect("PDU serializes");
+
+	assert_eq!(
+		value["msc4354_sticky"],
+		serde_json::json!({ "duration_ms": 300_000, "unknown": "key" })
+	);
+}
+
+#[test]
+fn out_of_range_sticky_object_is_clamped_not_rejected() {
+	let pdu = sticky_pdu(&serde_json::json!({ "duration_ms": 9_999_999_999_u64 }));
+	let sticky = pdu.sticky.as_deref().expect("sticky object is kept");
+
+	assert_eq!(super::sticky::duration_ms(sticky), Some(super::sticky::MAX_DURATION_MS));
+}
+
+#[test]
+fn nonsense_sticky_object_does_not_reject_the_pdu() {
+	let pdu = sticky_pdu(&serde_json::json!("nonsense"));
+	let sticky = pdu.sticky.as_deref().expect("sticky object is kept");
+
+	assert_eq!(super::sticky::duration_ms(sticky), None);
+}
+
+#[test]
+fn redaction_removes_stickiness() {
+	let mut pdu = sticky_pdu(&serde_json::json!({ "duration_ms": 300_000 }));
+	pdu.redact(&ruma::RoomVersionId::V11, serde_json::json!({}))
+		.expect("redaction succeeds");
+
+	assert!(pdu.sticky.is_none());
+}
+
+#[test]
+fn sticky_ttl_is_added_to_unsigned() {
+	let mut pdu = sticky_pdu(&serde_json::json!({ "duration_ms": 300_000 }));
+	pdu.origin_server_ts = ruma::MilliSecondsSinceUnixEpoch::now().get();
+	pdu.add_sticky_duration_ttl().expect("ttl is added");
+
+	let unsigned: serde_json::Value =
+		serde_json::from_str(pdu.unsigned.as_deref().expect("unsigned is set").get())
+			.expect("unsigned is valid");
+	let ttl = unsigned["msc4354_sticky_duration_ttl_ms"]
+		.as_u64()
+		.expect("ttl is a number");
+
+	assert!(ttl > 0 && ttl <= 300_000, "ttl {ttl} out of range");
+}
+
+#[test]
+fn no_ttl_for_events_that_are_not_sticky() {
+	let mut pdu = sticky_pdu(&serde_json::json!({ "duration_ms": "nonsense" }));
+	pdu.add_sticky_duration_ttl().expect("ttl is skipped");
+
+	assert!(pdu.unsigned.is_none());
+}
+
+#[test]
+fn set_unsigned_adds_the_sticky_ttl() {
+	// sticky events in the timeline get the hint through here, not the sticky
+	// section
+	let mut pdu = sticky_pdu(&serde_json::json!({ "duration_ms": 300_000 }));
+	pdu.origin_server_ts = ruma::MilliSecondsSinceUnixEpoch::now().get();
+	pdu.set_unsigned(None);
+
+	let unsigned: serde_json::Value =
+		serde_json::from_str(pdu.unsigned.as_deref().expect("unsigned is set").get())
+			.expect("unsigned is valid");
+
+	assert!(unsigned["msc4354_sticky_duration_ttl_ms"].is_number());
+}

src/core/matrix/pdu/unsigned.rs

@@ -3,7 +3,7 @@ use std::{borrow::Borrow, collections::BTreeMap};
 use ruma::MilliSecondsSinceUnixEpoch;
 use serde_json::value::{RawValue as RawJsonValue, Value as JsonValue, to_raw_value};
 
-use super::Pdu;
+use super::{Pdu, sticky};
 use crate::{Result, err, result::LogErr};
 
 impl Pdu {
@@ -17,6 +17,7 @@ impl Pdu {
 			self.remove_transaction_id().log_err().ok();
 		}
 		self.add_age().log_err().ok();
+		self.add_sticky_duration_ttl().log_err().ok();
 	}
 
 	pub fn remove_transaction_id(&mut self) -> Result {
@@ -58,6 +59,31 @@ impl Pdu {
 		Ok(())
 	}
 
+	pub fn add_sticky_duration_ttl(&mut self) -> Result {
+		use BTreeMap as Map;
+
+		let now = u64::from(MilliSecondsSinceUnixEpoch::now().get());
+		let Some(expires_at) = self
+			.sticky
+			.as_deref()
+			.and_then(|sticky| sticky::expires_at(self.origin_server_ts, sticky, now))
+		else {
+			return Ok(());
+		};
+
+		let mut unsigned: Map<&str, Box<RawJsonValue>> = self
+			.unsigned
+			.as_deref()
+			.map(RawJsonValue::get)
+			.map_or_else(|| Ok(Map::new()), serde_json::from_str)
+			.map_err(|e| err!(Database("Invalid unsigned in pdu event: {e}")))?;
+
+		unsigned.insert(sticky::TTL_UNSIGNED_KEY, to_raw_value(&expires_at.saturating_sub(now))?);
+		self.unsigned = Some(to_raw_value(&unsigned)?);
+
+		Ok(())
+	}
+
 	pub fn add_relation(&mut self, name: &str, pdu: Option<&Self>) -> Result {
 		use serde_json::Map;
 

src/core/matrix/state_res/benches.rs

@@ -414,6 +414,7 @@ where
 		content,
 		origin: None,
 		redacts: None,
+		sticky: None,
 		unsigned: None,
 		auth_events,
 		prev_events,

src/core/matrix/state_res/test_utils.rs

@@ -414,6 +414,7 @@ pub(crate) fn to_init_pdu_event(
 		content,
 		origin: None,
 		redacts: None,
+		sticky: None,
 		unsigned: None,
 		auth_events: vec![],
 		prev_events: vec![],
@@ -462,6 +463,7 @@ where
 		content,
 		origin: None,
 		redacts: None,
+		sticky: None,
 		unsigned: None,
 		auth_events,
 		prev_events,

src/service/rooms/pdu_metadata/bundled_aggregations.rs

@@ -281,6 +281,7 @@ mod tests {
 			origin_server_ts: UInt::try_from(1_234_567_890_u64).unwrap(),
 			kind: TimelineEventType::RoomMessage,
 			content: to_raw_value(&json!({"msgtype": "m.text", "body": "test"})).unwrap(),
+			sticky: None,
 			state_key: None,
 			prev_events: vec![],
 			depth: UInt::from(1_u32),
@@ -445,6 +446,7 @@ mod tests {
 			origin_server_ts: UInt::try_from(1_234_567_890_u64).unwrap(),
 			kind: event_type,
 			content: to_raw_value(&content).unwrap(),
+			sticky: None,
 			state_key: state_key.map(Into::into),
 			prev_events: vec![],
 			depth: UInt::from(1_u32),

src/service/rooms/timeline/append.rs

@@ -1,6 +1,7 @@
 use std::{
 	borrow::Borrow,
 	collections::{BTreeMap, HashMap, HashSet},
+	iter::once,
 	sync::Arc,
 };
 
@@ -18,17 +19,22 @@ use conduwuit::{
 };
 use conduwuit_core::{
 	err, error,
-	matrix::pdu::{PduCount, PduId, RawPduId},
-	utils::{self},
+	matrix::pdu::{PduCount, PduId, RawPduId, sticky},
+	result::LogErr,
+	utils::{self, stream::TryIgnore},
 };
-use futures::{FutureExt, StreamExt, TryFutureExt};
+use futures::{FutureExt, StreamExt, TryFutureExt, pin_mut};
 use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, RoomId,
-	RoomVersionId, UserId,
+	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, OwnedServerName,
+	RoomId, RoomVersionId, ServerName, UserId,
 	events::{
 		AnySyncTimelineEvent, GlobalAccountDataEventType, TimelineEventType,
 		push_rules::PushRulesEvent,
-		room::{encrypted::Relation, redaction::RoomRedactionEventContent},
+		room::{
+			encrypted::Relation,
+			member::{MembershipState, RoomMemberEventContent},
+			redaction::RoomRedactionEventContent,
+		},
 	},
 	push::{Action, Ruleset, Tweak},
 	room_version_rules::RoomVersionRules,
@@ -542,6 +548,10 @@ impl super::Service {
 					let target_user_id = UserId::parse(state_key)
 						.expect("This state_key was previously validated");
 
+					// must be checked before the membership update, which is what
+					// marks the server as being in the room
+					let joining_server = self.joining_server(pdu, &target_user_id, room_id).await;
+
 					// Update our membership info, we do this here incase a user is invited or
 					// knocked and immediately leaves we need the DB to record the invite or
 					// knock event for auth
@@ -549,6 +559,11 @@ impl super::Service {
 						.state_cache
 						.update_membership(room_id, &target_user_id, pdu, true)
 						.await?;
+
+					if let Some(server) = joining_server {
+						self.send_sticky_events_to_server(room_id, short_room_id, &server)
+							.await;
+					}
 				}
 			},
 			| TimelineEventType::RoomMessage => {
@@ -563,6 +578,80 @@ impl super::Service {
 		Ok(())
 	}
 
+	/// The server of `target_user_id`, if this membership event brings it into
+	/// the room for the first time.
+	async fn joining_server(
+		&self,
+		pdu: &PduEvent,
+		target_user_id: &UserId,
+		room_id: &RoomId,
+	) -> Option<OwnedServerName> {
+		let server = target_user_id.server_name();
+		if !self.services.config.allow_sticky_events
+			|| self.services.globals.server_is_ours(server)
+		{
+			return None;
+		}
+
+		let content: RoomMemberEventContent = pdu.get_content().ok()?;
+		if content.membership != MembershipState::Join {
+			return None;
+		}
+
+		let already_joined = self
+			.services
+			.state_cache
+			.server_in_room(server, room_id)
+			.await;
+
+		(!already_joined).then(|| server.to_owned())
+	}
+
+	/// MSC4354: a server which has just joined will not backfill far enough to
+	/// see our unexpired sticky events, so push them to it.
+	async fn send_sticky_events_to_server(
+		&self,
+		room_id: &RoomId,
+		short_room_id: ShortRoomId,
+		server: &ServerName,
+	) {
+		let now = utils::millis_since_unix_epoch();
+		let oldest_sticky_ts = now.saturating_sub(sticky::MAX_DURATION_MS);
+
+		let mut pdu_ids = Vec::new();
+		let pdus = self.pdus_rev(room_id, None).ignore_err();
+
+		pin_mut!(pdus);
+		while let Some((count, pdu)) = pdus.next().await {
+			if u64::from(pdu.origin_server_ts) < oldest_sticky_ts {
+				break;
+			}
+
+			let is_ours = self.services.globals.user_is_local(pdu.sender());
+			let is_sticky = pdu
+				.sticky
+				.as_deref()
+				.is_some_and(|sticky| sticky::is_sticky(pdu.origin_server_ts, sticky, now));
+
+			if is_ours && is_sticky {
+				pdu_ids.push(RawPduId::from(PduId {
+					shortroomid: short_room_id,
+					shorteventid: count,
+				}));
+			}
+		}
+
+		// oldest first, as MSC4354 asks for creation order
+		for pdu_id in pdu_ids.iter().rev() {
+			self.services
+				.sending
+				.send_pdu_servers(once(server.to_owned()).stream(), pdu_id)
+				.await
+				.log_err()
+				.ok();
+		}
+	}
+
 	/// Adds relation data to the incoming event and events it relates to.
 	async fn aggregate_relations(&self, pdu: &PduEvent, count2: Count) {
 		// CONCERN: If we receive events with a relation out-of-order, we never write

src/service/rooms/timeline/create.rs

@@ -88,6 +88,7 @@ impl super::Service {
 		let PartialPdu {
 			event_type,
 			content,
+			sticky,
 			unsigned,
 			state_key,
 			redacts,
@@ -209,6 +210,7 @@ impl super::Service {
 			),
 			kind: event_type,
 			content,
+			sticky,
 			state_key,
 			prev_events,
 			depth,

src/service/rooms/timeline/helpers.rs

@@ -1,6 +1,9 @@
 //! Helpers for submitting events with the right checks performed
 
-use conduwuit::{Err, Result, err, matrix::pdu::PartialPdu};
+use conduwuit::{
+	Err, Result, err,
+	matrix::pdu::{PartialPdu, sticky},
+};
 use ruma::{
 	MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, UserId,
 	events::{
@@ -12,6 +15,7 @@ use ruma::{
 			member::{MembershipState, RoomMemberEventContent},
 			server_acl::RoomServerAclEventContent,
 		},
+		sticky::StickyDurationMs,
 	},
 	serde::Raw,
 };
@@ -28,6 +32,7 @@ impl super::Service {
 		event_type: &StateEventType,
 		content: &Raw<AnyStateEventContent>,
 		state_key: &str,
+		sticky_duration_ms: Option<StickyDurationMs>,
 		timestamp: Option<MilliSecondsSinceUnixEpoch>,
 	) -> Result<OwnedEventId> {
 		let mut content: Raw<AnyStateEventContent> = content.clone();
@@ -42,6 +47,13 @@ impl super::Service {
 				PartialPdu {
 					event_type: event_type.to_string().into(),
 					content,
+					sticky: self
+						.services
+						.config
+						.allow_sticky_events
+						.then_some(sticky_duration_ms)
+						.flatten()
+						.map(sticky::object),
 					state_key: Some(state_key.into()),
 					timestamp,
 					..Default::default()

src/service/sending/sender.rs

@@ -14,6 +14,7 @@ use conduwuit::{
 };
 use conduwuit_core::{
 	Error, Event, Result, at, debug, err, error,
+	matrix::pdu::sticky,
 	result::LogErr,
 	utils::{
 		ReadyExt, calculate_hash, continue_exponential_backoff_secs,
@@ -268,7 +269,10 @@ impl Service {
 			}
 
 			let entry = txns.entry(dest.clone()).or_default();
-			if self.server.config.startup_netburst_keep >= 0 && entry.len() >= keep {
+			let over_budget =
+				self.server.config.startup_netburst_keep >= 0 && entry.len() >= keep;
+
+			if over_budget && !self.survives_netburst_trim(&event).await {
 				warn!(
 					startup_netburst_keep = self.server.config.startup_netburst_keep,
 					queue_size = entry.len(),
@@ -289,6 +293,29 @@ impl Service {
 		}
 	}
 
+	/// Whether this event skips the netburst budget, which MSC4354 asks of
+	/// still-sticky events. Only netburst delivers this queue, so with it
+	/// disabled there is nothing to keep the event for.
+	async fn survives_netburst_trim(&self, event: &SendingEvent) -> bool {
+		if !self.server.config.allow_sticky_events || !self.server.config.startup_netburst {
+			return false;
+		}
+
+		let SendingEvent::Pdu(pdu_id) = event else {
+			return false;
+		};
+
+		let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await else {
+			return false;
+		};
+
+		let now = MilliSecondsSinceUnixEpoch::now().get().into();
+
+		pdu.sticky
+			.as_deref()
+			.is_some_and(|sticky| sticky::is_sticky(pdu.origin_server_ts, sticky, now))
+	}
+
 	/// Selects any new events to send to the given destination.
 	#[tracing::instrument(
 		name = "select",