From 3c6680b5c73f9b95d3fe49ecdc9e363b5f54043e Mon Sep 17 00:00:00 2001 From: npilosov Date: Sun, 16 Aug 2026 06:45:36 +0300 Subject: [PATCH] feat: implement MSC4140 delayed events --- Cargo.toml | 2 + docs/calls/matrix_rtc.md | 8 + docs/development/compliance/msc.md | 9 +- src/api/client/capabilities.rs | 13 + src/api/client/delayed_events.rs | 215 +++++++++++++ src/api/client/message.rs | 3 + src/api/client/mod.rs | 2 + src/api/client/push/notifications.rs | 5 +- src/api/client/room/event.rs | 3 + src/api/client/state.rs | 5 +- src/api/client/versions.rs | 5 + src/api/router.rs | 25 +- src/core/config/mod.rs | 21 ++ src/core/matrix/pdu/format.rs | 1 + src/core/matrix/pdu/tests.rs | 28 ++ src/core/matrix/pdu/unsigned.rs | 3 +- src/database/maps.rs | 4 + src/service/delayed_events/mod.rs | 460 +++++++++++++++++++++++++++ src/service/mod.rs | 1 + src/service/services.rs | 7 +- tuwunel-example.toml | 13 + 21 files changed, 820 insertions(+), 13 deletions(-) create mode 100644 src/api/client/delayed_events.rs create mode 100644 src/service/delayed_events/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 22b9ae9d8f..741b03984c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -71,6 +71,7 @@ features = [ "http2", "json", "matched-path", + "original-uri", "tokio", "tracing", ] @@ -397,6 +398,7 @@ features = [ "unstable-msc4075", "unstable-msc4121", "unstable-msc4125", + "unstable-msc4140", "unstable-msc4143", "unstable-msc4186", "unstable-msc4195", diff --git a/docs/calls/matrix_rtc.md b/docs/calls/matrix_rtc.md index 36c4edc02e..c870caf3a1 100644 --- a/docs/calls/matrix_rtc.md +++ b/docs/calls/matrix_rtc.md @@ -8,6 +8,14 @@ supported here, paired with the [lk-jwt-service](https://github.com/element-hq/lk-jwt-service) which issues the access tokens clients use to join Livekit rooms. +Tuwunel enables MSC4140 delayed events by default. MatrixRTC uses them as a +heartbeat: if a client loses connectivity before it can leave a call cleanly, +the homeserver sends its previously scheduled leave event. The defaults allow +100 scheduled events per user and delays up to 24 hours. Operators can adjust +these limits with `max_delayed_events_per_user` and +`max_event_delay_duration`, or set either option to `0` to disable delayed +events. + This guide shows you how to deploy MatrixRTC/Element Call using Docker and Docker Compose, as Livekit only provides prebuilt Docker images. It is possible to run Livekit using their installation script, however this method diff --git a/docs/development/compliance/msc.md b/docs/development/compliance/msc.md index 5aa0dfc9ee..df4844649c 100644 --- a/docs/development/compliance/msc.md +++ b/docs/development/compliance/msc.md @@ -19,8 +19,8 @@ ## Counts - ✅ `yes`: 258 -- 🟨 `partial`: 34 -- ❌ `no`: 446 +- 🟨 `partial`: 35 +- ❌ `no`: 445 - ⬛ `n/a`: 286 ### Status by inventory bucket @@ -28,7 +28,7 @@ | Inv | yes | partial | no | n/a | total | |---|---|---|---|---|---| | merged | 186 | 7 | 7 | 59 | 259 | -| open | 64 | 26 | 399 | 175 | 664 | +| open | 64 | 27 | 398 | 175 | 664 | | closed | 8 | 1 | 40 | 52 | 101 | ## Merged @@ -443,7 +443,7 @@ in the [Out of scope](#out-of-scope) section. | MSC4145 | ❌ ● | 0/0 | Simple verified accounts | m.verified profile field and endpoint not implemented | | MSC4143 | ✅ ◐ | 80/80 | MatrixRTC | GET rtc/transports routed; only HS-side requirement of the MSC | | MSC4141 | ❌ ● | 0/0 | Time based notification filtering | time_and_day push rule condition not supported | -| MSC4140 | ❌ ● | 0/0 | Cancellable delayed events | delayed events endpoints not implemented despite Ruma types | +| MSC4140 | 🟨 ● | 75/85 | Cancellable delayed events | persistent scheduling, management endpoints, capability, and sender-private IDs; finalised retention and latest authenticated action-path draft pending | | MSC4136 | ❌ ● | 0/0 | Shared retry hints between servers | retry_hints in /send_join response not implemented | | MSC4128 | ✅ ● | 100/100 | Error on invalid auth where it is optional | invalid token returns error even on optional auth endpoints | | MSC4127 | ❌ ● | 0/0 | Removal of query string auth | removal of query string auth not implemented; still accepted | @@ -1112,4 +1112,3 @@ place of the (uniformly empty) `Correct/Impl` cell. | MSC688 | ⬛ ● | closed | Room Summaries (was: Calculate room names server-side) | stub Google doc; room summary work moved to heroes/MSC688 in spec | | MSC455 | ⬛ ● | closed | Do we want to specify a matrix:// URI scheme for rooms? (SPEC-5) | [→ MSC2312] stub Google doc; matrix:// URI scheme superseded by matrix: URI (... | | MSC441 | ⬛ ● | closed | Support for Reactions / Aggregations | [→ MSC2675/MSC2676] stub-only Google doc; superseded by MSC2675/MSC2676 react... | - diff --git a/src/api/client/capabilities.rs b/src/api/client/capabilities.rs index e3d9431c92..47e9710c4f 100644 --- a/src/api/client/capabilities.rs +++ b/src/api/client/capabilities.rs @@ -94,6 +94,19 @@ pub(crate) async fn get_capabilities_route( capabilities.set("im.nheko.msc3664.related_event_match", json!({"enabled": true}))?; } + // MSC4140: delayed events. + if services.config.max_event_delay_duration > 0 + && services.config.max_delayed_events_per_user > 0 + { + capabilities.set( + "org.matrix.msc4140.delayed_events", + json!({ + "max_delay_ms": services.config.max_event_delay_duration.saturating_mul(1000), + "max_scheduled": services.config.max_delayed_events_per_user, + }), + )?; + } + // MSC4323: advertise admin moderation only to admins; absence implies // neither suspend nor lock is available to the caller. if services diff --git a/src/api/client/delayed_events.rs b/src/api/client/delayed_events.rs new file mode 100644 index 0000000000..45a6641bcd --- /dev/null +++ b/src/api/client/delayed_events.rs @@ -0,0 +1,215 @@ +//! Client-server endpoints for MSC4140 delayed events. + +use axum::{ + extract::{OriginalUri, State}, + response::{IntoResponse, Response}, +}; +use ruma::{ + CanonicalJsonObject, + api::client::{ + delayed_events::{ + delayed_message_event, delayed_state_event, get_all_delayed_events, + get_delayed_event, send_delayed_event, update_delayed_event, + }, + message::send_message_event, + state::send_state_event, + }, +}; +use tuwunel_core::{Result, err}; +use tuwunel_service::delayed_events::ScheduleParams; + +use crate::{ClientIp, Ruma, RumaResponse}; + +fn parse_content(json: &str) -> Result { + serde_json::from_str(json) + .map_err(|error| err!(Request(BadJson("Invalid delayed event content: {error}")))) +} + +fn delay_from_query(uri: &http::Uri) -> Result> { + let Some(query) = uri.query() else { + return Ok(None); + }; + + query + .split('&') + .filter_map(|part| part.split_once('=')) + .find_map(|(key, value)| (key == "org.matrix.msc4140.delay").then_some(value)) + .map(|value| { + value + .parse::() + .map(std::time::Duration::from_millis) + .map_err(|_| err!(Request(InvalidParam("Invalid org.matrix.msc4140.delay.")))) + }) + .transpose() +} + +/// Dispatches the ordinary message-send path and the deprecated MSC4140 +/// query-parameter form, which intentionally share the same URL path. +pub(crate) async fn send_message_event_or_delayed_route( + State(services): State, + OriginalUri(uri): OriginalUri, + body: Ruma, +) -> Result { + if let Some(delay) = delay_from_query(&uri)? { + let delay_id = services + .delayed_events + .schedule(ScheduleParams { + user_id: body.sender_user(), + device_id: body.sender_device.as_deref(), + room_id: body.room_id.clone(), + event_type: body.event_type.clone().into(), + state_key: None, + content: parse_content(body.body.body.json().get())?, + txn_id: Some(body.txn_id.clone()), + delay, + }) + .await?; + + return Ok(RumaResponse(delayed_message_event::unstable::Response::new(delay_id)) + .into_response()); + } + + Ok(RumaResponse(super::send_message_event_route(State(services), body).await?) + .into_response()) +} + +/// Dispatches the ordinary state-send path and the deprecated MSC4140 +/// query-parameter form. +pub(crate) async fn send_state_event_or_delayed_route( + State(services): State, + OriginalUri(uri): OriginalUri, + body: Ruma, +) -> Result { + if let Some(delay) = delay_from_query(&uri)? { + let delay_id = services + .delayed_events + .schedule(ScheduleParams { + user_id: body.sender_user(), + device_id: body.sender_device.as_deref(), + room_id: body.room_id.clone(), + event_type: body.event_type.clone().into(), + state_key: Some(body.state_key.clone()), + content: parse_content(body.body.body.json().get())?, + txn_id: None, + delay, + }) + .await?; + + return Ok( + RumaResponse(delayed_state_event::unstable::Response::new(delay_id)).into_response() + ); + } + + Ok( + RumaResponse(super::send_state_event_for_key_route(State(services), body).await?) + .into_response(), + ) +} + +/// `PUT /_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/ +/// delayed_event/{event_type}/{txn_id}` +pub(crate) async fn send_delayed_event_route( + State(services): State, + body: Ruma, +) -> Result { + let delay_id = services + .delayed_events + .schedule(ScheduleParams { + user_id: body.sender_user(), + device_id: body.sender_device.as_deref(), + room_id: body.room_id.clone(), + event_type: body.event_type.clone(), + state_key: body.state_key.clone(), + content: parse_content(body.content.json().get())?, + txn_id: Some(body.txn_id.clone()), + delay: body.delay, + }) + .await?; + + Ok(send_delayed_event::unstable::Response::new(delay_id)) +} + +/// `POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}` +pub(crate) async fn update_delayed_event_v1_route( + State(services): State, + ClientIp(client): ClientIp, + body: Ruma, +) -> Result { + services + .delayed_events + .update(&body.delay_id, body.action.as_ref(), Some(body.sender_user()), client) + .await?; + + Ok(update_delayed_event::unstable_v1::Response::new()) +} + +/// `POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/ +/// {action}` +/// +/// This is the endpoint used by delegated LiveKit JWT services. MSC4140 makes +/// it intentionally unauthenticated; the service applies an IP rate limit. +pub(crate) async fn update_delayed_event_v2_route( + State(services): State, + ClientIp(client): ClientIp, + body: Ruma, +) -> Result { + services + .delayed_events + .update(&body.delay_id, body.action.as_ref(), None, client) + .await?; + + Ok(update_delayed_event::unstable_v2::Response::new()) +} + +/// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events` +pub(crate) async fn get_all_delayed_events_route( + State(services): State, + body: Ruma, +) -> Result { + Ok(get_all_delayed_events::unstable::Response::new( + services + .delayed_events + .list(body.sender_user()) + .await?, + )) +} + +/// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}` +pub(crate) async fn get_delayed_event_route( + State(services): State, + body: Ruma, +) -> Result { + Ok(get_delayed_event::unstable::Response::new( + services + .delayed_events + .get(&body.delay_id, body.sender_user()) + .await?, + )) +} + +#[cfg(test)] +mod tests { + use super::delay_from_query; + + #[test] + fn parses_legacy_delay_query() { + let uri = "/_matrix/client/v3/rooms/!room:example.org/send/m.room.message/tx?foo=bar&\ + org.matrix.msc4140.delay=123"; + assert_eq!( + delay_from_query(&uri.parse().unwrap()) + .unwrap() + .unwrap() + .as_millis(), + 123 + ); + } + + #[test] + fn ignores_requests_without_a_delay_query() { + assert!( + delay_from_query(&"/path".parse().unwrap()) + .unwrap() + .is_none() + ); + } +} diff --git a/src/api/client/message.rs b/src/api/client/message.rs index d98a26ad14..5d5e150f49 100644 --- a/src/api/client/message.rs +++ b/src/api/client/message.rs @@ -449,6 +449,9 @@ pub(crate) async fn with_membership( user_id: &UserId, encrypted: bool, ) -> PduEvent { + if pdu.sender() != user_id { + pdu.remove_transaction_id().log_err().ok(); + } annotate_membership(services, &mut pdu, user_id, encrypted).await; pdu } diff --git a/src/api/client/mod.rs b/src/api/client/mod.rs index 6ee5d051f6..e1a9fa194c 100644 --- a/src/api/client/mod.rs +++ b/src/api/client/mod.rs @@ -7,6 +7,7 @@ pub(super) mod backup; pub(super) mod capabilities; pub(super) mod context; pub(super) mod dehydrated_device; +pub(super) mod delayed_events; pub(super) mod device; pub(super) mod directory; pub(super) mod events; @@ -57,6 +58,7 @@ pub(super) use backup::*; pub(super) use capabilities::*; pub(super) use context::*; pub(super) use dehydrated_device::*; +pub(super) use delayed_events::*; pub(super) use device::*; pub(super) use directory::*; pub(super) use events::*; diff --git a/src/api/client/push/notifications.rs b/src/api/client/push/notifications.rs index 07b779b3f4..eb74a3dfab 100644 --- a/src/api/client/push/notifications.rs +++ b/src/api/client/push/notifications.rs @@ -63,12 +63,15 @@ pub(crate) async fn get_notifications_route( count: count.into(), }; - let event = services + let mut event = services .timeline .get_pdu_from_id(&pdu_id.into()) .await .ok() .filter(|event| !event.is_redacted())?; + if event.sender() != sender_user { + event.remove_transaction_id().ok(); + } let read = services .pusher diff --git a/src/api/client/room/event.rs b/src/api/client/room/event.rs index 835e52b555..bdb9dfdf90 100644 --- a/src/api/client/room/event.rs +++ b/src/api/client/room/event.rs @@ -85,6 +85,9 @@ pub(crate) async fn get_room_event_route( "sender": event.sender().as_str(), })); } + if event.sender() != sender_user { + event.remove_transaction_id().ok(); + } debug_assert!( event.event_id() == event_id && event.room_id() == room_id, diff --git a/src/api/client/state.rs b/src/api/client/state.rs index 0adc4d8e3c..ee70af8440 100644 --- a/src/api/client/state.rs +++ b/src/api/client/state.rs @@ -135,7 +135,7 @@ pub(crate) async fn get_state_events_for_key_route( )))); } - let event = services + let mut event = services .state_accessor .room_state_get(&body.room_id, &body.event_type, &body.state_key) .await @@ -146,6 +146,9 @@ pub(crate) async fn get_state_events_for_key_route( "Failed to get state event: {e}.", )))) })?; + if event.sender() != sender_user { + event.remove_transaction_id().ok(); + } let event_or_content = match body.format { | StateEventFormat::Event => json!({ diff --git a/src/api/client/versions.rs b/src/api/client/versions.rs index b949711593..0b3ce02a6e 100644 --- a/src/api/client/versions.rs +++ b/src/api/client/versions.rs @@ -43,6 +43,11 @@ pub(crate) async fn get_supported_versions_route( .rendezvous_enabled .then_some("org.matrix.msc4108"), ) + .chain( + (services.config.max_event_delay_duration > 0 + && services.config.max_delayed_events_per_user > 0) + .then_some("org.matrix.msc4140"), + ) .map(Into::into) .zip(once(true).cycle()) .collect(), diff --git a/src/api/router.rs b/src/api/router.rs index 0e30bf86fa..a22eecac54 100644 --- a/src/api/router.rs +++ b/src/api/router.rs @@ -9,7 +9,7 @@ pub mod state; use axum::{ Router, response::IntoResponse, - routing::{any, get, post}, + routing::{any, get, post, put}, }; pub use client_ip::{ConfiguredIpSource, TrustedPeerSubnets}; use http::{HeaderValue, header}; @@ -338,8 +338,23 @@ fn register_client_room_routes(router: Router) -> Router { fn register_client_state_and_sync_routes(router: Router) -> Router { router - .ruma_route(&client::send_message_event_route) - .ruma_route(&client::send_state_event_for_key_route) + .route( + "/_matrix/client/r0/rooms/{room_id}/send/{event_type}/{txn_id}", + put(client::send_message_event_or_delayed_route), + ) + .route( + "/_matrix/client/v3/rooms/{room_id}/send/{event_type}/{txn_id}", + put(client::send_message_event_or_delayed_route), + ) + .ruma_route(&client::send_delayed_event_route) + .route( + "/_matrix/client/r0/rooms/{room_id}/state/{event_type}/{state_key}", + put(client::send_state_event_or_delayed_route), + ) + .route( + "/_matrix/client/v3/rooms/{room_id}/state/{event_type}/{state_key}", + put(client::send_state_event_or_delayed_route), + ) .ruma_route(&client::get_state_events_route) .ruma_route(&client::get_state_events_for_key_route) // Ruma doesn't have support for multiple paths for a single endpoint yet, and these @@ -412,6 +427,10 @@ fn register_client_misc_routes(router: Router) -> Router { .ruma_route(&client::turn_server_route) .ruma_route(&client::get_transports_route) .ruma_route(&client::well_known_support) + .ruma_route(&client::get_all_delayed_events_route) + .ruma_route(&client::get_delayed_event_route) + .ruma_route(&client::update_delayed_event_v1_route) + .ruma_route(&client::update_delayed_event_v2_route) .ruma_route(&client::well_known_client) .ruma_route(&client::tuwunel_remote_version) .route("/_tuwunel/server_version", get(client::tuwunel_server_version)) diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index d524462bba..3ce2786c69 100644 --- a/src/core/config/mod.rs +++ b/src/core/config/mod.rs @@ -548,6 +548,23 @@ pub struct Config { )] pub max_request_size: usize, + /// Maximum delay for an MSC4140 delayed event, in seconds. The default is + /// 24 hours. + /// + /// Set to 0 to disable delayed events. + /// + /// default: 86400 + #[serde(default = "default_max_event_delay_duration")] + pub max_event_delay_duration: u64, + + /// Maximum number of MSC4140 delayed events a user may have scheduled. + /// + /// Set to 0 to disable delayed events. + /// + /// default: 100 + #[serde(default = "default_max_delayed_events_per_user")] + pub max_delayed_events_per_user: usize, + /// Maximum size of a response body buffered from a remote server. Applies /// to federation requests, push gateway and appservice transactions, and /// remote media fetched for URL previews. A peer cannot be trusted to honor @@ -5108,6 +5125,10 @@ fn default_ip_lookup_strategy() -> u8 { 5 } fn default_max_request_size() -> usize { 24 * 1024 * 1024 } +fn default_max_event_delay_duration() -> u64 { 24 * 60 * 60 } + +fn default_max_delayed_events_per_user() -> usize { 100 } + fn default_max_response_size() -> usize { 256 * 1024 * 1024 } fn default_max_pending_media_uploads() -> usize { 5 } diff --git a/src/core/matrix/pdu/format.rs b/src/core/matrix/pdu/format.rs index 0ab3781b96..8d7d234c23 100644 --- a/src/core/matrix/pdu/format.rs +++ b/src/core/matrix/pdu/format.rs @@ -22,6 +22,7 @@ pub fn into_outgoing_federation( .and_then(|val| val.as_object_mut()) { unsigned.remove("transaction_id"); + unsigned.remove("org.matrix.msc4140.delay_id"); } let Ok(room_rules) = room_version::rules(room_version) else { diff --git a/src/core/matrix/pdu/tests.rs b/src/core/matrix/pdu/tests.rs index 2878b05d44..6cea3609a7 100644 --- a/src/core/matrix/pdu/tests.rs +++ b/src/core/matrix/pdu/tests.rs @@ -165,6 +165,34 @@ fn remove_prev_state_absent_unsigned_noop() { assert!(pdu.unsigned.is_none()); } +#[test] +fn remove_transaction_id_strips_all_sender_private_ids() { + let mut pdu = member_pdu(&json!({ + "age": 4612, + "transaction_id": "txn", + "org.matrix.msc4140.delay_id": "delay", + })); + + pdu.remove_transaction_id().expect("strip failed"); + + let unsigned: serde_json::Value = serde_json::from_str( + pdu.unsigned + .as_ref() + .expect("unsigned kept") + .json() + .get(), + ) + .expect("valid unsigned"); + + assert!(unsigned.get("transaction_id").is_none()); + assert!( + unsigned + .get("org.matrix.msc4140.delay_id") + .is_none() + ); + assert_eq!(unsigned["age"], 4612); +} + fn replacement_raw() -> Raw { to_raw_value(&json!({ "type": "m.room.message", diff --git a/src/core/matrix/pdu/unsigned.rs b/src/core/matrix/pdu/unsigned.rs index a935c47edb..6a35e9d1a5 100644 --- a/src/core/matrix/pdu/unsigned.rs +++ b/src/core/matrix/pdu/unsigned.rs @@ -11,7 +11,7 @@ use serde_json::value::{RawValue as RawJsonValue, Value as JsonValue, to_raw_val use super::{Pdu, Unsigned}; use crate::{Result, err, implement, utils::BoolExt}; -/// Removes the local transaction ID from unsigned event metadata. +/// Removes sender-private IDs from unsigned event metadata. /// /// Other unsigned properties are retained and the object is re-encoded. An /// event without unsigned data is left unchanged. @@ -27,6 +27,7 @@ pub fn remove_transaction_id(&mut self) -> Result { .map_err(|e| err!(Database("Invalid unsigned in pdu event: {e}")))?; unsigned.remove("transaction_id"); + unsigned.remove("org.matrix.msc4140.delay_id"); self.unsigned = to_raw_value(&unsigned) .map(Into::into) .map(Some) diff --git a/src/database/maps.rs b/src/database/maps.rs index 9b3620139e..3a66b7dd69 100644 --- a/src/database/maps.rs +++ b/src/database/maps.rs @@ -102,6 +102,10 @@ pub(super) static MAPS: &[Descriptor] = &[ name: "bannedroomids", ..descriptor::RANDOM_SMALL }, + Descriptor { + name: "delayid_event", + ..descriptor::RANDOM_SMALL + }, Descriptor { name: "disabledroomids", ..descriptor::RANDOM_SMALL diff --git a/src/service/delayed_events/mod.rs b/src/service/delayed_events/mod.rs new file mode 100644 index 0000000000..41a48cd60d --- /dev/null +++ b/src/service/delayed_events/mod.rs @@ -0,0 +1,460 @@ +//! Persistent delayed-event scheduling for MSC4140. + +use std::{ + collections::HashMap, + net::IpAddr, + sync::Arc, + time::{Duration, Instant}, +}; + +use async_trait::async_trait; +use futures::TryStreamExt; +use http::StatusCode; +use ruma::{ + CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedRoomId, + OwnedTransactionId, OwnedUserId, UInt, + api::error::{ErrorKind, LimitExceededErrorData, RetryAfter}, + events::{AnyTimelineEventContent, TimelineEventType}, + serde::Raw, +}; +use serde::{Deserialize, Serialize}; +use tokio::sync::{Mutex, Notify}; +use tuwunel_core::{ + Err, Result, err, + matrix::pdu::PduBuilder, + utils::{rand::string_array, time::now_millis}, +}; +use tuwunel_database::{Deserialized, Json, Map}; + +const DELAY_ID_LENGTH: usize = 32; +const RATELIMITER_CAPACITY: usize = 4096; +const RATELIMITER_RATE: f64 = 1.0; +const RATELIMITER_BURST: f64 = 20.0; + +#[derive(Clone, Debug, Deserialize, Serialize)] +struct DelayedEvent { + delay_id: String, + user_id: OwnedUserId, + device_id: Option, + room_id: OwnedRoomId, + event_type: TimelineEventType, + state_key: Option, + content: CanonicalJsonObject, + txn_id: Option, + delay_ms: u64, + send_at: u64, + processing: bool, +} + +pub struct Service { + services: Arc, + delayid_event: Arc, + lock: Mutex<()>, + notify: Notify, + ratelimiter: std::sync::Mutex>, +} + +pub struct ScheduleParams<'a> { + pub user_id: &'a ruma::UserId, + pub device_id: Option<&'a ruma::DeviceId>, + pub room_id: OwnedRoomId, + pub event_type: TimelineEventType, + pub state_key: Option, + pub content: CanonicalJsonObject, + pub txn_id: Option, + pub delay: Duration, +} + +#[async_trait] +impl crate::Service for Service { + fn build(args: &crate::Args<'_>) -> Result> { + Ok(Arc::new(Self { + services: args.services.clone(), + delayid_event: args.db["delayid_event"].clone(), + lock: Mutex::new(()), + notify: Notify::new(), + ratelimiter: std::sync::Mutex::new(HashMap::new()), + })) + } + + async fn worker(self: Arc) -> Result { + self.recover_processing().await?; + + loop { + self.process_due().await?; + let wait = self.next_wait().await?; + + tokio::select! { + () = self.services.server.until_shutdown() => return Ok(()), + () = self.notify.notified() => {}, + () = tokio::time::sleep(wait) => {}, + } + } + } + + fn name(&self) -> &str { crate::service::make_name(std::module_path!()) } +} + +impl Service { + /// Schedule an event for later delivery and return its server-generated id. + pub async fn schedule(&self, params: ScheduleParams<'_>) -> Result { + let ScheduleParams { + user_id, + device_id, + room_id, + event_type, + state_key, + content, + txn_id, + delay, + } = params; + let max_delay = self + .services + .config + .max_event_delay_duration + .saturating_mul(1000); + let delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX); + + if max_delay == 0 || self.services.config.max_delayed_events_per_user == 0 { + return Err!(Request(Forbidden("Delayed events are disabled."))); + } + if delay_ms == 0 { + return Err!(Request(InvalidParam( + "The delayed event timeout must be greater than zero." + ))); + } + if delay_ms > max_delay { + return Err!(Request(Forbidden( + "The delayed event timeout exceeds the configured maximum." + ))); + } + + let _lock = self.lock.lock().await; + if let Some(txn_id) = txn_id.as_ref() + && let Ok(response) = self + .services + .transaction_ids + .existing_txnid(user_id, device_id, txn_id) + .await + { + return std::str::from_utf8(&response) + .ok() + .filter(|delay_id| delay_id.len() == DELAY_ID_LENGTH) + .map(ToOwned::to_owned) + .ok_or_else(|| { + err!(Request(InvalidParam( + "Tried to use txn_id already used for an incompatible endpoint." + ))) + }); + } + + let events = self.records().await?; + let scheduled = events + .iter() + .filter(|(_, event)| event.user_id == user_id) + .count(); + if scheduled >= self.services.config.max_delayed_events_per_user { + let now = now_millis(); + let retry_after = events + .iter() + .filter(|(_, event)| event.user_id == user_id) + .map(|(_, event)| event.send_at) + .min() + .map(|send_at| send_at.saturating_sub(now).div_ceil(1000).max(1)) + .map(Duration::from_secs) + .map(RetryAfter::Delay); + + return Err(tuwunel_core::Error::Request( + ErrorKind::LimitExceeded(LimitExceededErrorData { retry_after }), + "The maximum number of delayed events has been reached.".into(), + StatusCode::TOO_MANY_REQUESTS, + )); + } + + let delay_id = string_array::().to_string(); + let event = DelayedEvent { + delay_id: delay_id.clone(), + user_id: user_id.to_owned(), + device_id: device_id.map(ToOwned::to_owned), + room_id, + event_type, + state_key, + content, + txn_id: txn_id.clone(), + delay_ms, + send_at: now_millis().saturating_add(delay_ms), + processing: false, + }; + + self.delayid_event.put(&delay_id, Json(event)); + if let Some(txn_id) = txn_id.as_ref() { + self.services.transaction_ids.add_txnid( + user_id, + device_id, + txn_id, + delay_id.as_bytes(), + ); + } + self.notify.notify_one(); + Ok(delay_id) + } + + /// Update a scheduled event. Unauthenticated management requests are + /// rate-limited by client IP; authenticated requests are tied to the owner. + pub async fn update( + &self, + delay_id: &str, + action: &str, + owner: Option<&ruma::UserId>, + client: IpAddr, + ) -> Result { + if owner.is_none() { + self.check_rate_limit(client)?; + } + + let event = { + let _lock = self.lock.lock().await; + let mut event = self + .delayid_event + .get(delay_id) + .await + .deserialized::>() + .map(|Json(event)| event) + .map_err(|_| err!(Request(NotFound("Delayed event not found."))))?; + if owner.is_some_and(|owner| event.user_id != owner) { + return Err!(Request(NotFound("Delayed event not found."))); + } + + if event.processing { + return Err!(Request(NotFound("Delayed event is already being processed."))); + } + + match action { + | "cancel" => { + self.delayid_event.remove(&delay_id); + return Ok(()); + }, + | "restart" => { + event.send_at = now_millis().saturating_add(event.delay_ms); + self.delayid_event.put(delay_id, Json(event)); + self.notify.notify_one(); + return Ok(()); + }, + | "send" => { + event.processing = true; + self.delayid_event.put(delay_id, Json(&event)); + }, + | _ => return Err!(Request(InvalidParam("Unknown delayed event action."))), + } + + event + }; + + match self.send_event(&event).await { + | Ok(_) => { + self.delayid_event.remove(&delay_id); + Ok(()) + }, + | Err(error) => { + let mut event = event; + event.processing = false; + self.delayid_event.put(delay_id, Json(event)); + self.notify.notify_one(); + Err(error) + }, + } + } + + /// Return all scheduled events owned by a user. + pub async fn list( + &self, + user_id: &ruma::UserId, + ) -> Result> { + self.records() + .await? + .into_iter() + .filter(|(_, event)| event.user_id == user_id) + .map(|(_, event)| event_data(event)) + .collect() + } + + /// Return one scheduled event owned by a user. + pub async fn get( + &self, + delay_id: &str, + user_id: &ruma::UserId, + ) -> Result { + let event = self + .delayid_event + .get(delay_id) + .await + .deserialized::>() + .map(|Json(event)| event) + .map_err(|_| err!(Request(NotFound("Delayed event not found."))))?; + if event.user_id != user_id { + return Err!(Request(NotFound("Delayed event not found."))); + } + + event_data(event) + } + + async fn recover_processing(&self) -> Result { + let _lock = self.lock.lock().await; + for (delay_id, mut event) in self.records().await? { + if event.processing { + event.processing = false; + event.send_at = now_millis(); + self.delayid_event.put(&delay_id, Json(event)); + } + } + Ok(()) + } + + async fn process_due(&self) -> Result { + let now = now_millis(); + let due = { + let _lock = self.lock.lock().await; + let mut due = Vec::new(); + for (delay_id, mut event) in self.records().await? { + if event.send_at <= now && !event.processing { + event.processing = true; + self.delayid_event.put(&delay_id, Json(&event)); + due.push(event); + } + } + due + }; + + for event in due { + if let Err(error) = self.send_event(&event).await { + tracing::warn!(delay_id = %event.delay_id, ?error, "Failed to send delayed event"); + } + self.delayid_event.remove(&event.delay_id); + } + + Ok(()) + } + + async fn send_event(&self, event: &DelayedEvent) -> Result { + let state_lock = self + .services + .state + .mutex + .lock(&event.room_id) + .await; + let mut unsigned = std::collections::BTreeMap::new(); + unsigned.insert("org.matrix.msc4140.delay_id".to_owned(), event.delay_id.clone().into()); + if let Some(txn_id) = &event.txn_id { + unsigned.insert("transaction_id".to_owned(), txn_id.to_string().into()); + } + + let event_id = self + .services + .timeline + .build_and_append_pdu( + PduBuilder { + event_type: event.event_type.clone(), + content: Raw::new(&event.content)?, + state_key: event.state_key.clone().map(Into::into), + unsigned: Some(unsigned), + ..Default::default() + }, + &event.user_id, + &event.room_id, + &state_lock, + ) + .await?; + + Ok(event_id) + } + + async fn records(&self) -> Result> { + let mut records: Vec<(String, DelayedEvent)> = self + .delayid_event + .stream::<&str, Json>() + .map_ok(|(delay_id, Json(event))| (delay_id.to_owned(), event)) + .try_collect() + .await?; + + records.sort_unstable_by(|(left_id, left), (right_id, right)| { + left.send_at + .cmp(&right.send_at) + .then_with(|| left_id.cmp(right_id)) + }); + + Ok(records) + } + + async fn next_wait(&self) -> Result { + let now = now_millis(); + let next = self + .records() + .await? + .into_iter() + .filter(|(_, event)| !event.processing) + .map(|(_, event)| event.send_at) + .min(); + + Ok(Duration::from_millis( + next.map_or(60_000, |send_at| send_at.saturating_sub(now).max(1)), + )) + } + + fn check_rate_limit(&self, client: IpAddr) -> Result { + let now = Instant::now(); + let mut ratelimiter = self.ratelimiter.lock()?; + if ratelimiter.len() >= RATELIMITER_CAPACITY && !ratelimiter.contains_key(&client) { + ratelimiter.retain(|_, (last, tokens)| { + now.duration_since(*last) + .as_secs_f64() + .mul_add(RATELIMITER_RATE, *tokens) + < RATELIMITER_BURST + }); + if ratelimiter.len() >= RATELIMITER_CAPACITY { + return Err(tuwunel_core::Error::Request( + ErrorKind::LimitExceeded(LimitExceededErrorData { retry_after: None }), + "Too many delayed event actions.".into(), + StatusCode::TOO_MANY_REQUESTS, + )); + } + } + + let (last, tokens) = ratelimiter + .entry(client) + .or_insert((now, RATELIMITER_BURST)); + let available = now + .duration_since(*last) + .as_secs_f64() + .mul_add(RATELIMITER_RATE, *tokens) + .min(RATELIMITER_BURST); + if available < 1.0 { + return Err(tuwunel_core::Error::Request( + ErrorKind::LimitExceeded(LimitExceededErrorData { retry_after: None }), + "Too many delayed event actions.".into(), + StatusCode::TOO_MANY_REQUESTS, + )); + } + + *last = now; + *tokens = available - 1.0; + Ok(()) + } +} + +fn event_data( + event: DelayedEvent, +) -> Result { + let content: Raw = + Raw::from_json_string(serde_json::to_string(&event.content)?)?; + Ok(ruma::api::client::delayed_events::DelayedEventData::new( + event.delay_id, + event.room_id, + event.event_type, + event.state_key, + content, + Duration::from_millis(event.delay_ms), + MilliSecondsSinceUnixEpoch(UInt::new_saturating( + event.send_at.saturating_sub(event.delay_ms), + )), + )) +} diff --git a/src/service/mod.rs b/src/service/mod.rs index 60718c54ac..e129e6b7df 100644 --- a/src/service/mod.rs +++ b/src/service/mod.rs @@ -12,6 +12,7 @@ pub mod appservice; pub mod client; pub mod config; pub mod deactivate; +pub mod delayed_events; pub mod emergency; pub mod federation; pub mod fetcher; diff --git a/src/service/services.rs b/src/service/services.rs index fb625d7573..1ff9d2d36a 100644 --- a/src/service/services.rs +++ b/src/service/services.rs @@ -9,8 +9,8 @@ use tuwunel_database::Database; pub(crate) use crate::OnceServices; use crate::{ - account_data, admin, appservice, client, config, deactivate, emergency, federation, fetcher, - globals, key_backups, + account_data, admin, appservice, client, config, deactivate, delayed_events, emergency, + federation, fetcher, globals, key_backups, manager::Manager, media, membership, oauth, presence, profile, pusher, registration_tokens, rendezvous, resolver, @@ -64,6 +64,7 @@ pub struct Services { pub users: Arc, pub membership: Arc, pub deactivate: Arc, + pub delayed_events: Arc, pub oauth: Arc, pub retention: Arc, pub registration_tokens: Arc, @@ -131,6 +132,7 @@ pub async fn build(server: Arc) -> Result> { users: users::Service::build(&args)?, membership: membership::Service::build(&args)?, deactivate: deactivate::Service::build(&args)?, + delayed_events: delayed_events::Service::build(&args)?, oauth: oauth::Service::build(&args)?, retention: retention::Service::build(&args)?, registration_tokens: registration_tokens::Service::build(&args)?, @@ -199,6 +201,7 @@ pub(crate) fn services(&self) -> impl Iterator> + Send { cast!(self.users), cast!(self.membership), cast!(self.deactivate), + cast!(self.delayed_events), cast!(self.oauth), cast!(self.retention), cast!(self.registration_tokens), diff --git a/tuwunel-example.toml b/tuwunel-example.toml index dfec93204d..a6934853a4 100644 --- a/tuwunel-example.toml +++ b/tuwunel-example.toml @@ -411,6 +411,19 @@ # #max_request_size = 24 MiB +# Maximum delay for an MSC4140 delayed event, in seconds. The default is +# 24 hours. +# +# Set to 0 to disable delayed events. +# +#max_event_delay_duration = 86400 + +# Maximum number of MSC4140 delayed events a user may have scheduled. +# +# Set to 0 to disable delayed events. +# +#max_delayed_events_per_user = 100 + # Maximum size of a response body buffered from a remote server. Applies # to federation requests, push gateway and appservice transactions, and # remote media fetched for URL previews. A peer cannot be trusted to honor