From b38daf3310f97816d658d3c002e9dc55c9771a1a Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Sat, 24 Oct 2026 13:35:01 +0200 Subject: [PATCH] fix: preserve predecessors sharing a state-before ID Signed-off-by: Erwan Leboucher --- Cargo.lock | 2 + changelog.d/+state-before-forks.bugfix.md | 1 + src/service/Cargo.toml | 4 + .../rooms/event_handler/state_at_incoming.rs | 5 +- .../event_handler/state_at_incoming/tests.rs | 271 ++++++++++++++++++ 5 files changed, 282 insertions(+), 1 deletion(-) create mode 100644 changelog.d/+state-before-forks.bugfix.md create mode 100644 src/service/rooms/event_handler/state_at_incoming/tests.rs diff --git a/Cargo.lock b/Cargo.lock index d5e2615ef..5e9fc7351 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1122,6 +1122,7 @@ dependencies = [ "lru-cache", "nonzero_ext", "openidconnect", + "opentelemetry", "p256 0.14.0", "rand 0.10.3", "regex", @@ -1130,6 +1131,7 @@ dependencies = [ "resolvematrix", "ruma", "ruminuwuity", + "rustls", "rustyline-async", "sd-notify", "serde", diff --git a/changelog.d/+state-before-forks.bugfix.md b/changelog.d/+state-before-forks.bugfix.md new file mode 100644 index 000000000..70e3eadd7 --- /dev/null +++ b/changelog.d/+state-before-forks.bugfix.md @@ -0,0 +1 @@ +Fix incoming federation events losing state changes when predecessors share the same state-before snapshot. Contributed by @eleboucher. diff --git a/src/service/Cargo.toml b/src/service/Cargo.toml index f35a40aaa..41d1220fd 100644 --- a/src/service/Cargo.toml +++ b/src/service/Cargo.toml @@ -145,3 +145,7 @@ workspace = true [package.metadata.cargo-shear] ignored = ["reqwest_0_12"] + +[dev-dependencies] +opentelemetry.workspace = true +rustls = { workspace = true, features = ["ring"] } diff --git a/src/service/rooms/event_handler/state_at_incoming.rs b/src/service/rooms/event_handler/state_at_incoming.rs index e232df7bc..7025077c7 100644 --- a/src/service/rooms/event_handler/state_at_incoming.rs +++ b/src/service/rooms/event_handler/state_at_incoming.rs @@ -15,6 +15,9 @@ use ruma::{OwnedEventId, RoomId, room_version_rules::RoomVersionRules}; use crate::rooms::short::ShortStateHash; +#[cfg(test)] +mod tests; + impl super::Service { /// Resolves the state before the incoming event. /// @@ -124,7 +127,7 @@ impl super::Service { .pdu_shortstatehash(prev_eventid) .map_ok(move |sstatehash| (sstatehash, prev_event)) }) - .try_collect::>() + .try_collect::>() .inspect_err(|e| debug_error!("failed to calculate N-degree short state hashes: {e}")) .await else { diff --git a/src/service/rooms/event_handler/state_at_incoming/tests.rs b/src/service/rooms/event_handler/state_at_incoming/tests.rs new file mode 100644 index 000000000..8d4132568 --- /dev/null +++ b/src/service/rooms/event_handler/state_at_incoming/tests.rs @@ -0,0 +1,271 @@ +use std::sync::Arc; + +use conduwuit::{Config, PduEvent, Server, log::Log, metrics::Metrics}; +use futures::StreamExt; +use ruma::{RoomId, RoomVersionId, events::StateEventType}; +use serde_json::json; + +#[tokio::test] +async fn admitted_joins_do_not_disappear_after_a_merge() { + use conduwuit::matrix::PartialPdu; + use ruma::{ + UserId, + events::room::{ + join_rules::{JoinRule, RoomJoinRulesEventContent}, + member::{MembershipState, RoomMemberEventContent}, + }, + }; + + let (services, path) = test_services().await; + let alice = UserId::parse("@alice:test.example").unwrap(); + let bob = UserId::parse("@bob:bob.example").unwrap(); + let charlie = UserId::parse("@charlie:charlie.example").unwrap(); + let dave = UserId::parse("@dave:dave.example").unwrap(); + let mut remote_keys = Vec::new(); + for user in [&bob, &charlie, &dave] { + let key = ruma::signatures::Ed25519KeyPair::from_der( + &ruma::signatures::Ed25519KeyPair::generate(), + "test".into(), + ) + .unwrap(); + cache_key(&services, user.server_name(), &key); + remote_keys.push(key); + } + let bootstrap = services + .rooms + .state + .mutex + .lock("!bootstrap:test.example") + .await; + let create_id = services + .rooms + .timeline + .build_and_append_pdu( + PartialPdu { + event_type: "m.room.create".into(), + content: serde_json::value::to_raw_value(&json!({ "room_version": "12" })) + .unwrap(), + state_key: Some("".into()), + ..Default::default() + }, + &alice, + None, + &bootstrap, + ) + .await + .unwrap(); + drop(bootstrap); + let room = RoomId::parse(create_id.as_str().replacen('$', "!", 1)).unwrap(); + let lock = services.rooms.state.mutex.lock(&room).await; + services + .rooms + .timeline + .build_and_append_pdu( + PartialPdu::state( + alice.to_string(), + &RoomMemberEventContent::new(MembershipState::Join), + ), + &alice, + Some(&room), + &lock, + ) + .await + .unwrap(); + let rules_id = services + .rooms + .timeline + .build_and_append_pdu( + PartialPdu::state("", &RoomJoinRulesEventContent::new(JoinRule::Public)), + &alice, + Some(&room), + &lock, + ) + .await + .unwrap(); + drop(lock); + let rules = services.rooms.timeline.get_pdu(&rules_id).await.unwrap(); + let (dave_join, json) = signed_remote_event( + json!({ + "room_id": room, "sender": dave, "type": "m.room.member", "state_key": dave, + "content": { "membership": "join" }, "auth_events": [rules_id], "prev_events": [rules_id], + "depth": rules.depth + ruma::uint!(1), "origin_server_ts": conduwuit::utils::millis_since_unix_epoch(), + }), + dave.server_name(), + &remote_keys[2], + ); + assert!( + services + .rooms + .event_handler + .handle_incoming_pdu(dave.server_name(), &room, &dave_join.event_id, json, false) + .await + .unwrap() + .is_some() + ); + let mut joins = Vec::new(); + for (user, key) in [&bob, &charlie].into_iter().zip(&remote_keys) { + let signed = signed_remote_event( + json!({ + "room_id": room, "sender": user, "type": "m.room.member", "state_key": user, + "content": { "membership": "join" }, "auth_events": [rules_id], "prev_events": [dave_join.event_id], + "depth": dave_join.depth + ruma::uint!(1), "origin_server_ts": conduwuit::utils::millis_since_unix_epoch(), + }), + user.server_name(), + key, + ); + joins.push(signed); + } + for ((pdu, json), user) in joins.iter().zip([&bob, &charlie]) { + let accepted = services + .rooms + .event_handler + .handle_incoming_pdu(user.server_name(), &room, &pdu.event_id, json.clone(), false) + .await + .unwrap(); + assert!(accepted.is_some()); + } + let bob_hash = services + .rooms + .state_accessor + .pdu_shortstatehash(&joins[0].0.event_id) + .await + .unwrap(); + let charlie_hash = services + .rooms + .state_accessor + .pdu_shortstatehash(&joins[1].0.event_id) + .await + .unwrap(); + assert_eq!(bob_hash, charlie_hash); + let (merge, json) = signed_remote_event( + json!({ + "room_id": room, "sender": dave, "type": "m.room.message", + "content": { "msgtype": "m.text", "body": "merge" }, + "auth_events": [dave_join.event_id], + "prev_events": [joins[0].0.event_id, joins[1].0.event_id], + "depth": joins[0].0.depth + ruma::uint!(1), + "origin_server_ts": conduwuit::utils::millis_since_unix_epoch(), + }), + dave.server_name(), + &remote_keys[2], + ); + assert!( + services + .rooms + .event_handler + .handle_incoming_pdu(dave.server_name(), &room, &merge.event_id, json, false) + .await + .unwrap() + .is_some() + ); + let mut results = Vec::new(); + for ((user, key), (join, _)) in [&bob, &charlie].into_iter().zip(&remote_keys).zip(&joins) { + let membership = services + .rooms + .state_accessor + .room_state_get_content::( + &room, + &StateEventType::RoomMember, + user.as_str(), + ) + .await + .unwrap(); + assert_eq!(membership.membership, MembershipState::Join); + let prev_events = services + .rooms + .state + .get_forward_extremities(&room) + .collect::>() + .await; + let depth = services + .rooms + .timeline + .get_pdu(&prev_events[0]) + .await + .unwrap() + .depth + + ruma::uint!(1); + let (message, json) = signed_remote_event( + json!({ + "room_id": room, "sender": user, "type": "m.room.message", + "content": { "msgtype": "m.text", "body": format!("message from {user}") }, + "auth_events": [join.event_id], "prev_events": prev_events, + "depth": depth, "origin_server_ts": conduwuit::utils::millis_since_unix_epoch(), + }), + user.server_name(), + key, + ); + let result = services + .rooms + .event_handler + .handle_incoming_pdu(user.server_name(), &room, &message.event_id, json, false) + .await; + results.push(result.is_ok_and(|pdu| pdu.is_some())); + } + services.stop().await; + drop(services); + std::fs::remove_dir_all(path).unwrap(); + assert_eq!(results, [true, true], "both joined users must be able to send messages"); +} + +async fn test_services() -> (Arc, std::path::PathBuf) { + let _ = rustls::crypto::ring::default_provider().install_default(); + let path = std::env::temp_dir() + .join(format!("continuwuity-shared-state-{:016x}", rand::random::())); + std::fs::create_dir(&path).unwrap(); + let config: Config = serde_json::from_value(json!({ + "server_name": "test.example", + "database_path": path, + "db_cache_capacity_mb": 16.0, + "rocksdb_write_buffer_capacity_mb": 4.0, + "allow_local_presence": false + })) + .unwrap(); + let runtime = tokio::runtime::Handle::current(); + let server = Arc::new(Server::new( + config, + Some(runtime.clone()), + Log { + reload: Default::default(), + capture: Default::default(), + }, + Metrics::new(Some(runtime), opentelemetry::global::meter("test"), None), + )); + let services = crate::Services::build(server).await.unwrap(); + (services, path) +} + +fn cache_key( + services: &crate::Services, + server: &ruma::ServerName, + key: &ruma::signatures::Ed25519KeyPair, +) { + let mut response = ruma::api::federation::discovery::ServerSigningKeys::new( + server.to_owned(), + crate::server_keys::in_one_week(), + ); + let id: ruma::OwnedServerSigningKeyId = + format!("ed25519:{}", key.version()).try_into().unwrap(); + response.verify_keys.insert( + id.clone(), + ruma::api::federation::discovery::VerifyKey::new(ruma::serde::Base64::new( + key.public_key().to_vec(), + )), + ); + services.db["servernamekeyid_response"] + .put((server.as_str(), id.as_str()), database::Json(&response)); +} + +fn signed_remote_event( + value: serde_json::Value, + server: &ruma::ServerName, + key: &ruma::signatures::Ed25519KeyPair, +) -> (PduEvent, ruma::CanonicalJsonObject) { + let rules = RoomVersionId::V12.rules().unwrap(); + let mut value = serde_json::from_value(value).unwrap(); + ruma::signatures::hash_and_sign_event(server.as_str(), key, &mut value, &rules.redaction) + .unwrap(); + let id = conduwuit::matrix::event::gen_event_id(&value, &rules).unwrap(); + let pdu = PduEvent::from_id_val(&id, value.clone()).unwrap(); + (pdu, value) +}