From 43197f695c6039c5c5a37cd0e35d182d3ad791d8 Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Sun, 2 Aug 2026 10:14:17 +0200 Subject: [PATCH] feat(sync): Add MSC4354 sticky events --- Cargo.toml | 1 + changelog.d/+cf65f70c.feature.md | 1 + conduwuit-example.toml | 6 + src/api/client/room/create.rs | 1 + src/api/client/send.rs | 12 +- src/api/client/state.rs | 1 + src/api/client/sync/v3/joined.rs | 85 +++++++++- src/api/client/sync/v3/left.rs | 1 + src/api/client/unversioned.rs | 15 +- src/core/config/mod.rs | 8 + src/core/matrix/event.rs | 4 + src/core/matrix/event/format.rs | 10 ++ src/core/matrix/pdu.rs | 12 ++ src/core/matrix/pdu/partial.rs | 3 + src/core/matrix/pdu/redact.rs | 3 + src/core/matrix/pdu/sticky.rs | 155 ++++++++++++++++++ src/core/matrix/pdu/tests.rs | 93 +++++++++++ src/core/matrix/pdu/unsigned.rs | 28 +++- src/core/matrix/state_res/benches.rs | 1 + src/core/matrix/state_res/test_utils.rs | 2 + .../pdu_metadata/bundled_aggregations.rs | 2 + src/service/rooms/timeline/append.rs | 101 +++++++++++- src/service/rooms/timeline/create.rs | 2 + src/service/rooms/timeline/helpers.rs | 14 +- src/service/sending/sender.rs | 29 +++- 25 files changed, 572 insertions(+), 18 deletions(-) create mode 100644 changelog.d/+cf65f70c.feature.md create mode 100644 src/core/matrix/pdu/sticky.rs diff --git a/Cargo.toml b/Cargo.toml index 2f33a9dc2..7fe0906e8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -377,6 +377,7 @@ features = [ "unstable-msc4293", "unstable-msc4406", "unstable-msc4439", + "unstable-msc4354", "unstable-msc4466", "unstable-extensible-events", ] diff --git a/changelog.d/+cf65f70c.feature.md b/changelog.d/+cf65f70c.feature.md new file mode 100644 index 000000000..106fa8e30 --- /dev/null +++ b/changelog.d/+cf65f70c.feature.md @@ -0,0 +1 @@ +Added MSC4354 sticky events, behind the `allow_sticky_events` config option. Contributed by @eleboucher. diff --git a/conduwuit-example.toml b/conduwuit-example.toml index f8b062306..aa105b1fc 100644 --- a/conduwuit-example.toml +++ b/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 diff --git a/src/api/client/room/create.rs b/src/api/client/room/create.rs index 6d7fd269b..157acf72a 100644 --- a/src/api/client/room/create.rs +++ b/src/api/client/room/create.rs @@ -541,6 +541,7 @@ pub(crate) async fn create_room_route( &content, &state_key, None, + None, ) .boxed() .await diff --git a/src/api/client/send.rs b/src/api/client/send.rs index 277b6b428..ff07073f0 100644 --- a/src/api/client/send.rs +++ b/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 diff --git a/src/api/client/state.rs b/src/api/client/state.rs index cd9a491cc..60d382202 100644 --- a/src/api/client/state.rs +++ b/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 { diff --git a/src/api/client/sync/v3/joined.rs b/src/api/client/sync/v3/joined.rs index ba46a0091..653f878ee 100644 --- a/src/api/client/sync/v3/joined.rs +++ b/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, timeline: Timeline, + sticky: Sticky, summary: Option, notification_counts: Option, 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, diff --git a/src/api/client/sync/v3/left.rs b/src/api/client/sync/v3/left.rs index 83077a9df..c0f22e48f 100644 --- a/src/api/client/sync/v3/left.rs +++ b/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 diff --git a/src/api/client/unversioned.rs b/src/api/client/unversioned.rs index 8c2c63c60..8f5233aaf 100644 --- a/src/api/client/unversioned.rs +++ b/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, _body: Ruma, ) -> Result { - 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` diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index 663a90c0b..bc768ab62 100644 --- a/src/core/config/mod.rs +++ b/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 diff --git a/src/core/matrix/event.rs b/src/core/matrix/event.rs index 5af35634c..1b44b4ac1 100644 --- a/src/core/matrix/event.rs +++ b/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; diff --git a/src/core/matrix/event/format.rs b/src/core/matrix/event/format.rs index 32c4cf06b..c650f01ce 100644 --- a/src/core/matrix/event/format.rs +++ b/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(pub(super) E); @@ -36,6 +37,9 @@ impl<'a, E: Event> From> for Raw { 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> for Raw { 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> for Raw { 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); } diff --git a/src/core/matrix/pdu.rs b/src/core/matrix/pdu.rs index b05389988..8d8a855d7 100644 --- a/src/core/matrix/pdu.rs +++ b/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, + /// 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>, + #[serde(skip_serializing_if = "Option::is_none")] pub state_key: Option, @@ -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 } diff --git a/src/core/matrix/pdu/partial.rs b/src/core/matrix/pdu/partial.rs index 059726873..e31845ce7 100644 --- a/src/core/matrix/pdu/partial.rs +++ b/src/core/matrix/pdu/partial.rs @@ -20,6 +20,8 @@ pub struct PartialPdu { pub content: Box, + pub sticky: Option>, + pub unsigned: Option, pub state_key: Option, @@ -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, diff --git a/src/core/matrix/pdu/redact.rs b/src/core/matrix/pdu/redact.rs index e60dccdd5..ffee4fd49 100644 --- a/src/core/matrix/pdu/redact.rs +++ b/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}"))) })?; diff --git a/src/core/matrix/pdu/sticky.rs b/src/core/matrix/pdu/sticky.rs new file mode 100644 index 000000000..fc07bed7d --- /dev/null +++ b/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 { + 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 { + serde_json::from_str::(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 { + 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 { 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)); + } +} diff --git a/src/core/matrix/pdu/tests.rs b/src/core/matrix/pdu/tests.rs index ed9b7caaa..d5af4f9d4 100644 --- a/src/core/matrix/pdu/tests.rs +++ b/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()); +} diff --git a/src/core/matrix/pdu/unsigned.rs b/src/core/matrix/pdu/unsigned.rs index be293a408..eff5a8247 100644 --- a/src/core/matrix/pdu/unsigned.rs +++ b/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> = 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; diff --git a/src/core/matrix/state_res/benches.rs b/src/core/matrix/state_res/benches.rs index 992f9cf6b..f400cc244 100644 --- a/src/core/matrix/state_res/benches.rs +++ b/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, diff --git a/src/core/matrix/state_res/test_utils.rs b/src/core/matrix/state_res/test_utils.rs index 8cadd4815..3bd0c001a 100644 --- a/src/core/matrix/state_res/test_utils.rs +++ b/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, diff --git a/src/service/rooms/pdu_metadata/bundled_aggregations.rs b/src/service/rooms/pdu_metadata/bundled_aggregations.rs index 059d5a64a..46e496b8d 100644 --- a/src/service/rooms/pdu_metadata/bundled_aggregations.rs +++ b/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), diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 6730f3fd2..05577af95 100644 --- a/src/service/rooms/timeline/append.rs +++ b/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 { + 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 diff --git a/src/service/rooms/timeline/create.rs b/src/service/rooms/timeline/create.rs index 60fc9e42d..6387f2b16 100644 --- a/src/service/rooms/timeline/create.rs +++ b/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, diff --git a/src/service/rooms/timeline/helpers.rs b/src/service/rooms/timeline/helpers.rs index c025dc69a..50384bba6 100644 --- a/src/service/rooms/timeline/helpers.rs +++ b/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, state_key: &str, + sticky_duration_ms: Option, timestamp: Option, ) -> Result { let mut content: Raw = 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() diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index c07139841..c4703ba43 100644 --- a/src/service/sending/sender.rs +++ b/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",