chore: Reformat with latest nightly

This commit is contained in:
Ginger committed 2026-09-01 12:55:05 -04:00
1 parent 03d669594e
commit 9fbabe2210
113 files changed
+981 -831

No files matched your search

+9 -7
View File
@@ -206,8 +206,9 @@ impl crate::Context<'_> {
let first_seen = scheduled.insert(auth_event_id.clone());
let auth_pdu = if let Some(auth_pdu) = cached_events.get(&auth_event_id) {
// NOTE: events might be referenced multiple times (like the create event)
// so this saves some cheeky db lookup time
// NOTE: events might be referenced multiple times (like the
// create event) so this saves some cheeky db lookup
// time
Some(auth_pdu.clone())
} else if first_seen {
match self.services.rooms.timeline.get_pdu(&auth_event_id).await {
@@ -221,9 +222,10 @@ impl crate::Context<'_> {
None
};
// NOTE: Depth is used as the primary sorting key here, even though it has no
// bearing on state resolution or anything. Timestamp is used as a
// tiebreaker, failing back to lexicographical comparison.
// NOTE: Depth is used as the primary sorting key here, even
// though it has no bearing on state resolution or anything.
// Timestamp is used as a tiebreaker, failing back to
// lexicographical comparison.
let (depth, ts) = auth_pdu
.as_ref()
.map_or((UInt::MAX, UInt::MAX), |pdu| (pdu.depth, pdu.origin_server_ts));
@@ -254,8 +256,8 @@ impl crate::Context<'_> {
// We have this PDU so will want to traverse it.
stack.push(child_pdu);
} else {
// We don't have this PDU locally so we can't traverse its auth events,
// but we can still render it as a node.
// We don't have this PDU locally so we can't traverse its
// auth events, but we can still render it as a node.
render_node(
&mut graph,
&child.node_id,
+4 -2
View File
@@ -361,7 +361,8 @@ impl crate::Context<'_> {
.fetch_remote_content(&mxc, None, server.as_deref(), timeout)
.await?;
// Grab the length of the content before clearing it to not flood the output
// Grab the length of the content before clearing it to not flood the
// output
let len = result.content.as_ref().expect("content").len();
result.content.as_mut().expect("content").clear();
@@ -386,7 +387,8 @@ impl crate::Context<'_> {
.fetch_remote_thumbnail(&mxc, None, server.as_deref(), timeout, &dim)
.await?;
// Grab the length of the content before clearing it to not flood the output
// Grab the length of the content before clearing it to not flood the
// output
let len = result.content.as_ref().expect("content").len();
result.content.as_mut().expect("content").clear();
+2 -1
View File
@@ -14,7 +14,8 @@ impl crate::Context<'_> {
include_empty: bool,
no_details: bool,
) -> Result {
// TODO: i know there's a way to do this with clap, but i can't seem to find it
// TODO: i know there's a way to do this with clap, but i can't seem to
// find it
let page = page.unwrap_or(1);
let mut rooms = self
.services
+2 -1
View File
@@ -37,7 +37,8 @@ pub(super) async fn process(command: RoomDirectoryCommand, context: &Context<'_>
context.write_str("Room unpublished").await
},
| RoomDirectoryCommand::List { page } => {
// TODO: i know there's a way to do this with clap, but i can't seem to find it
// TODO: i know there's a way to do this with clap, but i can't seem
// to find it
let page = page.unwrap_or(1);
let mut rooms: Vec<_> = services
.rooms
+3 -2
View File
@@ -28,8 +28,9 @@ impl crate::Context<'_> {
}
pub(super) async fn reload_config(&self, path: Option<PathBuf>) -> Result {
// The path argument is only what's optionally passed via the admin command,
// so we need to merge it with the existing paths if any were given at startup.
// The path argument is only what's optionally passed via the admin
// command, so we need to merge it with the existing paths if any were
// given at startup.
let mut paths = Vec::new();
// Add previously saved paths to the argument list
+6 -6
View File
@@ -83,8 +83,8 @@ pub(crate) async fn change_password_route(
body: Ruma<change_password::v3::Request>,
) -> Result<change_password::v3::Response> {
let identity = if let Some(identity) = body.identity.as_ref() {
// A signed-in user is trying to change their password, prompt them for their
// existing one
// A signed-in user is trying to change their password, prompt them for
// their existing one
services
.uiaa
@@ -99,10 +99,10 @@ pub(crate) async fn change_password_route(
)
.await?
} else {
// A signed-out user is trying to reset their password, prompt them for email
// confirmation. Note that we do not _send_ an email here, their client should
// have already hit `/account/password/requestToken` to send the email. We
// just validate it.
// A signed-out user is trying to reset their password, prompt them for
// email confirmation. Note that we do not _send_ an email here, their
// client should have already hit `/account/password/requestToken` to
// send the email. We just validate it.
services
.uiaa
+5 -4
View File
@@ -41,8 +41,8 @@ pub(crate) async fn register_route(
return Err!(Request(GuestAccessForbidden("Guests may not register on this server.")));
}
// Allow registration if it's enabled in the config file or if this is the first
// run (so the first user account can be created)
// Allow registration if it's enabled in the config file or if this is the
// first run (so the first user account can be created)
let allow_registration =
services.config.allow_registration || services.firstrun.is_first_run();
@@ -108,8 +108,9 @@ pub(crate) async fn register_route(
};
let (token, device) = if !body.inhibit_login {
// If UIAA is disabled, we can't create a device. In that case only appservices
// can reach this point in the first place, so we return an error for them.
// If UIAA is disabled, we can't create a device. In that case only
// appservices can reach this point in the first place, so we return
// an error for them.
if !services.config.oauth.compatibility_mode().uiaa_available() {
return Err!(Request(AppserviceLoginUnsupported(
"User-interactive appservice registration is not available on this server."
+2 -2
View File
@@ -247,8 +247,8 @@ pub(crate) async fn add_backup_keys_for_session_route(
| Ordering::Less => true,
| Ordering::Greater => false,
| Ordering::Equal => {
// If both have same `first_message_index`, prefer the one with lower
// `forwarded_count`
// If both have same `first_message_index`, prefer the one
// with lower `forwarded_count`
let old_forwarded_count = old_key
.get_field::<UInt>("forwarded_count")?
.unwrap_or(UInt::MAX);
+5 -3
View File
@@ -111,8 +111,9 @@ pub(crate) async fn upload_keys_route(
.await
.and_then(|keys| keys.deserialize().map_err(Into::into))
{
// NOTE: also serves as a workaround for a nheko bug which omits cross-signing
// NOTE: signatures when re-uploading the same DeviceKeys.
// NOTE: also serves as a workaround for a nheko bug which omits
// cross-signing NOTE: signatures when re-uploading the same
// DeviceKeys.
if existing_keys.keys == deser_device_keys.keys {
debug!(
%sender_user,
@@ -273,7 +274,8 @@ async fn uiaa_needed_to_upload_keys(
);
if let Some(existing_master_signing_key) = existing_master_signing_key {
// If a master key exists, UIAA is required if any of the keys are different.
// If a master key exists, UIAA is required if any of the keys are
// different.
master_signing_key != Some(existing_master_signing_key)
|| user_signing_key != existing_user_signing_key
+5 -4
View File
@@ -77,8 +77,9 @@ pub(crate) async fn invite_user_route(
.await;
// ignored invites aren't handled here
// since the recipient's membership should still be changed to `invite`.
// they're filtered out in the individual /sync handlers.
// since the recipient's membership should still be changed to
// `invite`. they're filtered out in the individual /sync
// handlers.
if matches!(recipient_filter_level, FilterLevel::Block) {
return Err!(Request(InviteBlocked(
"{recipient_user} has blocked invites from you."
@@ -190,8 +191,8 @@ pub(crate) async fn invite_helper(
.send_federation_request(recipient_user.server_name(), request)
.await?;
// We do not add the event_id field to the pdu here because of signature and
// hashes checks
// We do not add the event_id field to the pdu here because of signature
// and hashes checks
let (event_id, value) = gen_event_id_canonical_json(
&response.event,
&room_version_id
+18 -17
View File
@@ -190,10 +190,10 @@ async fn knock_room_by_id_helper(
}
}
// For knock_restricted rooms, check if the user meets the restricted conditions
// If they do, attempt to join instead of knock
// This is not mentioned in the spec, but should be allowable (we're allowed to
// auto-join invites to knocked rooms)
// For knock_restricted rooms, check if the user meets the restricted
// conditions If they do, attempt to join instead of knock
// This is not mentioned in the spec, but should be allowable (we're allowed
// to auto-join invites to knocked rooms)
let join_rule = services.rooms.state_accessor.get_join_rules(room_id).await;
if let JoinRule::KnockRestricted(restricted) = &join_rule {
@@ -226,9 +226,9 @@ async fn knock_room_by_id_helper(
"{sender_user} meets the restricted criteria in knock_restricted room \
{room_id}, attempting to join instead of knock"
);
// For this case, we need to drop the state lock and get a new one in
// join_room_by_id_helper We need to release the lock here and let
// join_room_by_id_helper acquire it again
// For this case, we need to drop the state lock and get a new one
// in join_room_by_id_helper We need to release the lock here and
// let join_room_by_id_helper acquire it again
drop(state_lock);
match services
.rooms
@@ -401,8 +401,8 @@ async fn knock_room_helper_local(
to_canonical_value(content).expect("event is valid, we just created it"),
);
// In order to create a compatible ref hash (EventID) the `hashes` field needs
// to be present
// In order to create a compatible ref hash (EventID) the `hashes` field
// needs to be present
services
.server_keys
.hash_and_sign_event(&mut knock_event_stub, &room_version_rules)?;
@@ -446,10 +446,11 @@ async fn knock_room_helper_local(
.map_err(|e| err!(BadServerResponse("Invalid knock event PDU: {e:?}")))?;
info!("Updating membership locally to knock state with provided stripped state events");
// TODO: this call does not appear to do anything because `update_membership`
// doesn't call `mark_as_knock`. investigate further, ideally with the aim of
// removing this call entirely -- Ginger thinks `update_membership` should only
// be called from `force_state` and `append_pdu`.
// TODO: this call does not appear to do anything because
// `update_membership` doesn't call `mark_as_knock`. investigate further,
// ideally with the aim of removing this call entirely -- Ginger thinks
// `update_membership` should only be called from `force_state` and
// `append_pdu`.
services
.rooms
.state_cache
@@ -524,8 +525,8 @@ async fn knock_room_helper_remote(
to_canonical_value(knock_content).expect("event is valid, we just created it"),
);
// In order to create a compatible ref hash (EventID) the `hashes` field needs
// to be present
// In order to create a compatible ref hash (EventID) the `hashes` field
// needs to be present
services
.server_keys
.hash_and_sign_event(&mut knock_event_stub, &room_version_rules)?;
@@ -671,8 +672,8 @@ async fn knock_room_helper_remote(
.await?;
info!("Setting final room state for new room");
// We set the room state after inserting the pdu, so that we never have a moment
// in time where events in the current room state do not exist
// We set the room state after inserting the pdu, so that we never have a
// moment in time where events in the current room state do not exist
services
.rooms
.state
+19 -15
View File
@@ -110,16 +110,16 @@ pub async fn leave_room(
otherwise `build_and_append_pdu` will take care of updating the state cache for us.
*/
// `leave_pdu` is the outlier `m.room.member` event which will be synced to the
// user. if it's None the sync handler will create a dummy PDU.
// `leave_pdu` is the outlier `m.room.member` event which will be synced to
// the user. if it's None the sync handler will create a dummy PDU.
let leave_pdu = if is_banned.or(is_disabled).await {
// case 1: the room is banned/disabled. we don't want to federate with another
// server to leave, so we can't create an outlier PDU.
// case 1: the room is banned/disabled. we don't want to federate with
// another server to leave, so we can't create an outlier PDU.
None
} else if dont_have_room.and(not_knocked).await {
// case 2: ask a remote server to assist us with leaving
// we always mark the room as left locally, regardless of if the federated leave
// failed
// we always mark the room as left locally, regardless of if the
// federated leave failed
remote_leave_room(services, user_id, room_id, reason.clone(), HashSet::new())
.await
@@ -159,12 +159,14 @@ pub async fn leave_room(
)
.await?;
// `build_and_append_pdu` calls `mark_as_left` internally, so we return early.
// `build_and_append_pdu` calls `mark_as_left` internally, so we
// return early.
return Ok(());
},
| Err(_) => {
// an exception to case 3 is if the user isn't even in the room they're trying
// to leave. this can happen if the client's caching is wrong.
// an exception to case 3 is if the user isn't even in the room
// they're trying to leave. this can happen if the client's
// caching is wrong.
debug_warn!(
"Trying to leave a room you are not a member of, marking room as left \
locally."
@@ -177,11 +179,13 @@ pub async fn leave_room(
.left_state(user_id, room_id)
.await
.inspect_err(|err| {
// `left_state` may return an Err if the user _is_ in the room they're
// trying to leave, but the membership cache is incorrect and
// `left_state` may return an Err if the user _is_ in
// the room they're trying to leave, but the
// membership cache is incorrect and
// they're cached as being joined. In this situation
// we save a `None` to the `roomuserid_leftcount` table, which generates
// and sends a dummy leave to the client.
// we save a `None` to the `roomuserid_leftcount` table,
// which generates and sends a dummy leave to the
// client.
warn!(
?err,
"Trying to leave room not cached as leave, sending dummy leave \
@@ -370,8 +374,8 @@ pub async fn remote_leave_room<S: ::std::hash::BuildHasher>(
// room v3 and above removed the "event_id" field from remote PDU format
leave_event_stub.remove("event_id");
// In order to create a compatible ref hash (EventID) the `hashes` field needs
// to be present
// In order to create a compatible ref hash (EventID) the `hashes` field
// needs to be present
services
.server_keys
.hash_and_sign_event(&mut leave_event_stub, &room_version_rules)?;
+4 -4
View File
@@ -40,8 +40,8 @@ pub(crate) async fn get_pushrules_all_route(
.and_then(|event| event.get("content").cloned())
.filter(CanonicalJsonValue::is_object)
else {
// user somehow has non-existent push rule event. recreate it and return server
// default silently
// user somehow has non-existent push rule event. recreate it and return
// server default silently
return recreate_push_rules_and_return(&services, sender_user).await;
};
@@ -111,8 +111,8 @@ pub(crate) async fn get_pushrules_global_route(
.and_then(|event| event.get("content").cloned())
.filter(CanonicalJsonValue::is_object)
else {
// user somehow has non-existent push rule event. recreate it and return server
// default silently
// user somehow has non-existent push rule event. recreate it and return
// server default silently
let global_ruleset = Ruleset::server_default(sender_user);
let event = PushRulesEvent::new(PushRulesEventContent::new(global_ruleset.clone()));
+2 -2
View File
@@ -48,8 +48,8 @@ pub(crate) async fn report_room_route(
delay_response().await;
// We log this early in case the room ID does actually exist, in which case
// admins who scan their logs can see the report and choose to investigate at
// their discretion.
// admins who scan their logs can see the report and choose to investigate
// at their discretion.
info!(
"Received room report by user {sender_user} for room {} with reason: \"{}\"",
body.room_id, body.reason
+13 -13
View File
@@ -90,15 +90,15 @@ pub(crate) async fn create_room_route(
};
let room_version_rules = room_version.rules().unwrap();
// For custom room IDs, if the user is creating a room with a v1 room ID format,
// we can just use that ID directly. However, if it's a custom *v2* room ID, we
// need to make sure that we don't generate one, which would in turn trick us
// into generating invalid v2 room events.
// For custom room IDs, if the user is creating a room with a v1 room ID
// format, we can just use that ID directly. However, if it's a custom *v2*
// room ID, we need to make sure that we don't generate one, which would in
// turn trick us into generating invalid v2 room events.
//
// expect_room_id is the custom room ID that the user is expecting - for v2
// formatted rooms, we check that the m.room.create event's generated room ID
// exactly matches this, and abort if it doesn't. Otherwise, we use it as the
// room ID itself.
// formatted rooms, we check that the m.room.create event's generated room
// ID exactly matches this, and abort if it doesn't. Otherwise, we use it
// as the room ID itself.
let expect_room_id = {
let body_ref = body.json_body.as_ref().unwrap();
if let Some(CanonicalJsonValue::String(room_id)) = body_ref
@@ -165,10 +165,10 @@ pub(crate) async fn create_room_route(
continue;
}
// if the recipient of the invite is local and has the sender blocked, error
// out. if the recipient is remote we can't tell yet, and if they're local and
// have the sender _ignored_ their invite will be filtered out in
// the handlers for the individual /sync endpoints
// if the recipient of the invite is local and has the sender blocked,
// error out. if the recipient is remote we can't tell yet, and if
// they're local and have the sender _ignored_ their invite will be
// filtered out in the handlers for the individual /sync endpoints
if services.globals.user_is_local(recipient_user)
&& matches!(
services
@@ -652,8 +652,8 @@ fn default_power_levels_content(
power_levels_content["events"]["m.poll.response"] =
serde_json::to_value(0).expect("0 is valid Value");
// synapse does this too. clients do not expose these permissions. it prevents
// default users from calling public rooms, for obvious reasons.
// synapse does this too. clients do not expose these permissions. it
// prevents default users from calling public rooms, for obvious reasons.
if *visibility == room::Visibility::Public {
power_levels_content["events"]["m.call.invite"] =
serde_json::to_value(50).expect("50 is valid Value");
+33 -25
View File
@@ -77,8 +77,8 @@ async fn update_parents(
continue; // Skip updating rooms the sender isn't in.
}
let state_lock = services.rooms.state.mutex.lock(parent_id.as_str()).await;
// We're now fetching state from the *space* that has the old room as a *child*.
// Follow along. This will be on the test.
// We're now fetching state from the *space* that has the old room as a
// *child*. Follow along. This will be on the test.
let Ok(child) = services
.rooms
.state_accessor
@@ -98,13 +98,15 @@ async fn update_parents(
);
})
else {
// If the space does not have a child event for this room, we can skip it
// If the space does not have a child event for this room, we can
// skip it
continue;
};
// ...the upgrading server SHOULD send a new m.space.child event with state_key
// set to the new room's ID, copying the order and suggested fields from the
// content of the m.space.child with state_key of the previous room ID.
// ...the upgrading server SHOULD send a new m.space.child event with
// state_key set to the new room's ID, copying the order and suggested
// fields from the content of the m.space.child with state_key of the
// previous room ID.
services
.rooms
.timeline
@@ -199,7 +201,8 @@ async fn update_children(
);
})
else {
// If the child does not have a parent event for this room, we can skip it.
// If the child does not have a parent event for this room, we can
// skip it.
continue;
};
@@ -228,9 +231,10 @@ async fn update_children(
))
.ok();
// If the previous m.space.parent event has canonical set to true in content,
// homeservers SHOULD update the old state event to set canonical to false,
// while setting it to true in the newly-sent m.space.parent event.
// If the previous m.space.parent event has canonical set to true in
// content, homeservers SHOULD update the old state event to set
// canonical to false, while setting it to true in the newly-sent
// m.space.parent event.
if parent.canonical {
services
.rooms
@@ -301,11 +305,12 @@ pub(crate) async fn upgrade_room_route(
return Err!(Request(Forbidden("Upgrading the admin room this way is not allowed.")));
}
// 1. Check that the user has permission to send m.room.tombstone events in the
// room.
// 1. Check that the user has permission to send m.room.tombstone events in
// the room.
let old_room_state_lock = services.rooms.state.mutex.lock(body.room_id.as_str()).await;
// Check tombstone permission by attempting to create (but not send) the event.
// Check tombstone permission by attempting to create (but not send) the
// event.
services
.rooms
.timeline
@@ -389,9 +394,9 @@ pub(crate) async fn upgrade_room_route(
let new_room_state_lock = if let Some(new_room_id) = replacement_room_id.as_ref() {
services.rooms.state.mutex.lock(new_room_id.as_str()).await
} else {
// NOTE: Using a hardcoded room ID for the temporary mutex means only one room
// can be created at a time. This is actually beneficial, as it reduces the
// risk of concurrent in-flight collisions.
// NOTE: Using a hardcoded room ID for the temporary mutex means only
// one room can be created at a time. This is actually beneficial, as
// it reduces the risk of concurrent in-flight collisions.
services.rooms.state.mutex.lock("!new-room").await
};
debug!("Upgrading {} to room version {}", &body.room_id, &body.new_version);
@@ -408,9 +413,9 @@ pub(crate) async fn upgrade_room_route(
.await?;
drop(new_room_state_lock);
// re-acquire a new lock with the new room ID.
// We don't actually need a state lock for sending the m.room.create event, but
// we get one anyway because the function requires it and I can't be bothered
// refactoring it.
// We don't actually need a state lock for sending the m.room.create event,
// but we get one anyway because the function requires it and I can't be
// bothered refactoring it.
let (replacement_room_id, new_room_state_lock) =
if new_version_rules.room_id_format == RoomIdFormatVersion::V2 {
let parsed_room_id = RoomId::new_v2(
@@ -470,8 +475,9 @@ pub(crate) async fn upgrade_room_route(
| Ok(v) => v.content().to_owned(),
| Err(_) => continue, // Skipping missing events.
};
// If this is a power levels event, and the new room version has creators,
// we need to make sure they dont appear in the users block of power levels.
// If this is a power levels event, and the new room version has
// creators, we need to make sure they dont appear in the users
// block of power levels.
if *event_type == StateEventType::RoomPowerLevels {
let creators = body
.additional_creators
@@ -550,8 +556,8 @@ pub(crate) async fn upgrade_room_route(
)?;
}
// 5. Send a `m.room.tombstone` event to the old room to indicate that it is not
// intended to be used any further.
// 5. Send a `m.room.tombstone` event to the old room to indicate that it is
// not intended to be used any further.
debug!(target=?body.room_id, "Sending tombstone to old room");
services
.rooms
@@ -577,7 +583,8 @@ pub(crate) async fn upgrade_room_route(
.get_room_power_levels(&body.room_id)
.await;
// Setting events_default and invite to the greater of 50 and users_default + 1
// Setting events_default and invite to the greater of 50 and users_default
// + 1
let new_level = max(
int!(50),
power_levels
@@ -591,7 +598,8 @@ pub(crate) async fn upgrade_room_route(
power_levels.events_default = new_level;
power_levels.invite = new_level;
// 6. Modify the power levels in the old room to prevent sending of events and
// 6. Modify the power levels in the old room to prevent sending of events
// and
// inviting new users
// Spec dictates that this is allowed to fail.
debug!(target=?body.room_id, ?new_level, "Raising power level in old room to lock it");
+2 -1
View File
@@ -222,7 +222,8 @@ pub(crate) async fn login_route(
.await?
};
// send client well-known if specified so the client knows to reconfigure itself
// send client well-known if specified so the client knows to reconfigure
// itself
let client_discovery_info: Option<DiscoveryInfo> = services
.server
.config
+3 -3
View File
@@ -15,9 +15,9 @@ pub(crate) async fn get_hierarchy_route(
State(services): State<crate::State>,
body: Ruma<get_hierarchy::v1::Request>,
) -> Result<get_hierarchy::v1::Response> {
// We don't do pagination for this route (and therefore ignore `limit`), since
// there's no reasonable way to handle a space hierarchy changing during
// pagination.
// We don't do pagination for this route (and therefore ignore `limit`),
// since there's no reasonable way to handle a space hierarchy changing
// during pagination.
// Default to MAX_MAX_DEPTH when the client doesn't specify one, so an
// unbounded traversal can never happen.
+9 -7
View File
@@ -67,14 +67,16 @@ async fn load_timeline(
})?;
if last_timeline_count <= starting_count {
// no messages have been sent in this room since `starting_count`
// no messages have been sent in this room since
// `starting_count`
return Ok(TimelinePdus::default());
}
// for incremental sync, stream from the DB all PDUs which were sent after
// `starting_count` but before `ending_count`, including `ending_count` but
// not `starting_count`. this code is pretty similar to the initial sync
// branch, they're separate to allow for future optimization
// for incremental sync, stream from the DB all PDUs which were sent
// after `starting_count` but before `ending_count`, including
// `ending_count` but not `starting_count`. this code is pretty
// similar to the initial sync branch, they're separate to allow
// for future optimization
services
.rooms
.timeline
@@ -99,8 +101,8 @@ async fn load_timeline(
.boxed()
},
| None => {
// For initial sync, stream from the DB all PDUs before and including
// `ending_count` in reverse order
// For initial sync, stream from the DB all PDUs before and
// including `ending_count` in reverse order
services
.rooms
.timeline
+37 -32
View File
@@ -148,10 +148,11 @@ async fn build_ephemeral(
SyncContext { syncing_user, last_sync_end_count, .. }: SyncContext<'_>,
room_id: &RoomId,
) -> Result<Ephemeral> {
// note: some of the futures below are boxed. this is because, without the box,
// rustc produces over thirty inscrutable errors in `mod.rs` at the call-site
// of `load_joined_room`. I don't know why boxing them fixes this -- it seems
// to be related to the async closures and borrowing from the sync context.
// note: some of the futures below are boxed. this is because, without the
// box, rustc produces over thirty inscrutable errors in `mod.rs` at the
// call-site of `load_joined_room`. I don't know why boxing them fixes this
// -- it seems to be related to the async closures and borrowing from the
// sync context.
// collect updates to read receipts
let receipt_events = services
@@ -176,8 +177,8 @@ async fn build_ephemeral(
| Some(last_sync_end_count) => {
match services.rooms.typing.last_typing_update(room_id).await {
| Ok(last_typing_update) => {
// update the typing list if the users typing have changed since the last
// sync
// update the typing list if the users typing have
// changed since the last sync
last_typing_update > last_sync_end_count
},
| Err(err) => {
@@ -288,7 +289,8 @@ async fn build_sticky_events(
.flatten()
.map(PduCount::Normal);
// nothing older than this can still be sticky, so the walk back never goes far
// 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();
@@ -393,14 +395,14 @@ async fn build_state_and_timeline(
// history
let prev_batch = timeline.pdus.front().map(at!(0));
// note: we always indicate a limited timeline if the syncing user just joined
// the room, to indicate to the client that it should request backfill (and to
// copy Synapse's behavior). for federated room joins, the `timeline` will
// usually only include the syncing user's join event.
// note: we always indicate a limited timeline if the syncing user just
// joined the room, to indicate to the client that it should request
// backfill (and to copy Synapse's behavior). for federated room joins, the
// `timeline` will usually only include the syncing user's join event.
let limited = timeline.limited || joined_since_last_sync;
// filter out ignored events from the timeline and convert the PDUs into Ruma's
// AnySyncTimelineEvent type
// filter out ignored events from the timeline and convert the PDUs into
// Ruma's AnySyncTimelineEvent type
let filtered_timeline = timeline
.pdus
.into_iter()
@@ -442,8 +444,8 @@ async fn fetch_shortstatehashes(
room_id: &RoomId,
) -> Result<ShortStateHashes> {
// the room state currently.
// TODO: this should be the room state as of `current_count`, but there's no way
// to get that right now.
// TODO: this should be the room state as of `current_count`, but there's no
// way to get that right now.
let current_shortstatehash = services
.rooms
.state
@@ -484,7 +486,8 @@ async fn fetch_shortstatehashes(
}
},
| None => {
// No events have been sent since the last sync, or we just joined this room
// No events have been sent since the last sync, or we just
// joined this room
None
},
}
@@ -561,12 +564,13 @@ async fn build_state_events(
} = shortstatehashes;
if timeline.pdus.is_empty() {
// If the timeline is empty there can't possibly be any changes to the state
// If the timeline is empty there can't possibly be any changes to the
// state
return Ok(vec![]);
}
// the user IDs of members whose membership needs to be sent to the client, if
// lazy-loading is enabled.
// the user IDs of members whose membership needs to be sent to the client,
// if lazy-loading is enabled.
let lazily_loaded_members =
prepare_lazily_loaded_members(services, sync_context, room_id, timeline.senders()).await;
@@ -619,14 +623,14 @@ async fn build_notification_counts(
) -> Result<Option<UnreadNotificationsCount>> {
// determine whether to actually update the notification counts
let should_send_notification_counts = async {
// if we're going to sync some timeline events, the notification count has
// definitely changed to include them
// if we're going to sync some timeline events, the notification count
// has definitely changed to include them
if !timeline.pdus.is_empty() {
return true;
}
// if this is an initial sync, we need to send notification counts because the
// client doesn't know what they are yet
// if this is an initial sync, we need to send notification counts
// because the client doesn't know what they are yet
let Some(last_sync_end_count) = last_sync_end_count else {
return true;
};
@@ -637,8 +641,8 @@ async fn build_notification_counts(
.last_notification_read(syncing_user, room_id)
.await;
// if the syncing user has read the events we sent during the last sync, we need
// to send a new notification count on this sync.
// if the syncing user has read the events we sent during the last sync,
// we need to send a new notification count on this sync.
if last_notification_read > last_sync_end_count {
return true;
}
@@ -683,8 +687,8 @@ async fn check_joined_since_last_sync(
SyncContext { syncing_user, .. }: SyncContext<'_>,
) -> Result<bool> {
let Some(last_sync_end_shortstatehash) = last_sync_end_shortstatehash else {
// For initial syncs always return false, since there's no "last sync" for the
// user to have joined since.
// For initial syncs always return false, since there's no "last sync"
// for the user to have joined since.
return Ok(false);
};
@@ -703,8 +707,8 @@ async fn check_joined_since_last_sync(
// TODO: If the requesting user got state-reset out of the room, this
// will be `true` when it shouldn't be. this function should never be called
// in that situation, but it may be if the membership cache didn't get updated.
// the root cause of this needs to be addressed
// in that situation, but it may be if the membership cache didn't get
// updated. the root cause of this needs to be addressed
let joined_since_last_sync =
membership_during_previous_sync.is_none_or(|content: RoomMemberEventContent| {
content.membership != MembershipState::Join
@@ -729,7 +733,8 @@ async fn build_room_summary(
state_events: &[PduEvent],
joined_since_last_sync: bool,
) -> Result<Option<RoomSummary>> {
// determine whether any events in the state or timeline are membership events.
// determine whether any events in the state or timeline are membership
// events.
let are_syncing_membership_events = timeline
.pdus
.iter()
@@ -863,8 +868,8 @@ async fn build_device_list_updates(
.state_get(current_shortstatehash, &StateEventType::RoomEncryption, "")
.is_ok();
// initial syncs don't include device updates, and rooms which aren't encrypted
// don't affect them, so return early in either of those cases
// initial syncs don't include device updates, and rooms which aren't
// encrypted don't affect them, so return early in either of those cases
if last_sync_end_count.is_none() || !(is_encrypted_room.await) {
return Ok(DeviceListUpdates::new());
}
+16 -14
View File
@@ -59,22 +59,23 @@ pub(super) async fn load_left_room(
.await
.ok()
else {
// if we get here, the membership cache is incorrect, likely due to a state
// reset
// if we get here, the membership cache is incorrect, likely due to a
// state reset
debug_warn!("attempting to sync left room but no left count exists");
return Ok(None);
};
// return early if we haven't gotten to this leave yet.
// this can happen if the user leaves while a sync response is being generated
// this can happen if the user leaves while a sync response is being
// generated
if current_count < left_count {
return Ok(None);
}
// return early if:
// - this is an initial sync and the room filter doesn't include leaves, or
// - this is an incremental sync, and we've already synced the leave, and the
// room filter doesn't include leaves
// - this is an incremental sync, and we've already synced the leave, and
// the room filter doesn't include leaves
if last_sync_end_count.is_none_or(|last_sync_end_count| last_sync_end_count >= left_count)
&& !filter.room.include_leave
{
@@ -110,8 +111,8 @@ pub(super) async fn load_left_room(
(TimelinePdus::default(), vec![leave_membership_event])
},
| Some(leave_membership_event) => {
// we have this room in our DB, and can fetch the state and timeline from when
// the user left.
// we have this room in our DB, and can fetch the state and timeline
// from when the user left.
let leave_state_key = syncing_user;
debug_assert_eq!(
@@ -120,9 +121,9 @@ pub(super) async fn load_left_room(
"leave PDU should be for the user requesting the sync"
);
// the shortstatehash of the state _immediately before_ the syncing user left
// this room. the state represented here _does not_ include
// `leave_membership_event`.
// the shortstatehash of the state _immediately before_ the syncing
// user left this room. the state represented here _does not_
// include `leave_membership_event`.
let leave_shortstatehash = services
.rooms
.state_accessor
@@ -246,9 +247,9 @@ async fn build_left_state_and_timeline(
)
.await?;
// the timeline index can lag behind the leave PDU, so `load_timeline` may omit
// it. we already hold the leave event, so ensure it's always synced to the
// client.
// the timeline index can lag behind the leave PDU, so `load_timeline` may
// omit it. we already hold the leave event, so ensure it's always synced
// to the client.
let leave_synced = timeline
.pdus
.iter()
@@ -294,7 +295,8 @@ async fn build_left_state_and_timeline(
[1]: https://github.com/matrix-org/matrix-js-sdk/issues/5071
*/
// `state` should only ever include one membership event for the syncing user
// `state` should only ever include one membership event for the syncing
// user
let membership_event_index = state.iter().position(|pdu| {
*pdu.event_type() == TimelineEventType::RoomMember
&& pdu.state_key() == Some(syncing_user.as_str())
+7 -6
View File
@@ -41,7 +41,8 @@ pub(super) async fn build_state_initial(
let event_ids_in_timeline: HashSet<_> =
timeline.pdus.iter().map(|pdu| &pdu.1.event_id).collect();
// load the keys and event IDs of the state events at the start of the timeline
// load the keys and event IDs of the state events at the start of the
// timeline
let (shortstatekeys, event_ids): (Vec<_>, Vec<_>) = services
.rooms
.state_accessor
@@ -167,11 +168,11 @@ pub(super) async fn build_state_incremental<'a>(
.await;
if !use_state_after {
// If state_after isn't enabled, filter out state events which also exist
// in the timeline. If splits exist in the DAG, this may not be exactly the same
// thing as the state diff ending at the start of the timeline, but Synapse
// also does this and it's technically more useful behavior anyway.
// See: https://github.com/element-hq/synapse/issues/16941
// If state_after isn't enabled, filter out state events which also
// exist in the timeline. If splits exist in the DAG, this may not be
// exactly the same thing as the state diff ending at the start of the
// timeline, but Synapse also does this and it's technically more
// useful behavior anyway. See: https://github.com/element-hq/synapse/issues/16941
for (_, pdu) in &timeline.pdus {
state_event_ids.remove(pdu.event_id());
+12 -10
View File
@@ -205,8 +205,8 @@ async fn collect_sync_response(
globalsince: u64,
known_rooms: &KnownRooms,
) -> Result<SyncCollection> {
// Bounds every read below and becomes the `pos` returned to the client. Zero
// would be echoed back as a request for an initial sync.
// Bounds every read below and becomes the `pos` returned to the client.
// Zero would be echoed back as a request for an initial sync.
let next_batch = services.globals.current_count()?.max(1);
let all_joined_rooms = services
@@ -1132,8 +1132,8 @@ async fn collect_sticky_events(
.iter()
.filter(|room_id| joined_room_ids.contains(*room_id))
{
// a room the client has not seen before needs its whole backlog, which a
// single stream position cannot express
// a room the client has not seen before needs its whole backlog, which
// a single stream position cannot express
let initial = todo_rooms
.get(room_id)
.is_some_and(|&(_, _, roomsince)| roomsince == 0);
@@ -1462,7 +1462,8 @@ where
.await
.is_ok();
// Skip the state diff only: `room_keys_changed` below must run for every room.
// Skip the state diff only: `room_keys_changed` below must run for
// every room.
if let Some(since_shortstatehash) =
since_shortstatehash.filter(|since| *since != current_shortstatehash)
{
@@ -1533,8 +1534,8 @@ where
}
},
| MembershipState::Leave => {
// Write down users that have left encrypted rooms we
// are in
// Write down users that have left
// encrypted rooms we are in
left_encrypted_users.insert(user_id.clone());
},
| _ => {},
@@ -1544,7 +1545,8 @@ where
}
}
if joined_since_last_sync || new_encrypted_room {
// If the user is in a new encrypted room, give them all joined users
// If the user is in a new encrypted room, give them all
// joined users
device_list_changes.extend(
services
.rooms
@@ -1580,8 +1582,8 @@ where
let dont_share_encrypted_room =
!share_encrypted_room(services, sender_user, &user_id, None).await;
// If the user doesn't share an encrypted room with the target anymore, we need
// to tell them
// If the user doesn't share an encrypted room with the target anymore,
// we need to tell them
if dont_share_encrypted_room {
device_list_left.insert(user_id);
}
+2 -2
View File
@@ -62,8 +62,8 @@ where
let body = {
let max_body_size = services.server.config.max_request_size;
// Check if the Content-Length header is present and valid, saves us streaming
// the response into memory
// Check if the Content-Length header is present and valid, saves us
// streaming the response into memory
if let Some(content_length) = parts.headers.get(http::header::CONTENT_LENGTH) {
if let Ok(content_length) = content_length
.to_str()
+2 -2
View File
@@ -85,8 +85,8 @@ pub(crate) async fn get_backfill_route(
pdu.remove_transaction_id().log_err().ok();
// Add age, as this is specified
pdu.add_age().log_err().ok();
// It's not clear if we should strip or add any more data, leave as is.
// In particular: Redaction?
// It's not clear if we should strip or add any more data, leave as
// is. In particular: Redaction?
Ok(pdu)
})
.try_filter_map(|pdu| async move {
+11 -11
View File
@@ -75,8 +75,8 @@ pub(crate) async fn create_invite_route(
.expect("must have create event in invite state by this point");
// We can now perform the banned remote server check with the create event.
// N.B. this checks the sender field, which is technically incorrect for rooms
// v10 and below. This usually isn't the case though so sue me
// N.B. this checks the sender field, which is technically incorrect for
// rooms v10 and below. This usually isn't the case though so sue me
let creator = create_event_json
.get("sender")
.and_then(|v| v.as_str())
@@ -122,8 +122,8 @@ pub(crate) async fn create_invite_route(
return Err!(Request(Forbidden("Invite rejected by antispam service.")));
}
// If we're already in the room, ensure that neither the origin nor ourselves
// are ACL'd.
// If we're already in the room, ensure that neither the origin nor
// ourselves are ACL'd.
let resident = services
.rooms
.state_cache
@@ -159,15 +159,15 @@ pub(crate) async fn create_invite_route(
// If we are active in the room, the remote server will notify us about the
// join/invite through /send. If we are not in the room, we need to manually
// record the invited state for client /sync through update_membership(), and
// send the invite PDU to the relevant appservices.
// record the invited state for client /sync through update_membership(),
// and send the invite PDU to the relevant appservices.
if !resident {
// We will start by recording the room's create event as an outlier.
// This will allow us to recognise it later in case the sender revokes the
// invite over federation later. We could store more state from the invite
// request, but we will get that during send_join anyway.
// This is safe to just add directly as an outlier as we already auth checked it
// during validation.
// This will allow us to recognise it later in case the sender revokes
// the invite over federation later. We could store more state from
// the invite request, but we will get that during send_join anyway.
// This is safe to just add directly as an outlier as we already auth
// checked it during validation.
services
.rooms
.outlier
+13 -11
View File
@@ -99,8 +99,8 @@ pub(crate) async fn create_join_event_template_route(
);
let join_authorized_via_users_server: Option<OwnedUserId> = {
if is_joined || is_invited {
// User is already joined or invited and consequently does not need an
// authorising user
// User is already joined or invited and consequently does not need
// an authorising user
None
} else if !room_version_rules.authorization.restricted_join_rule {
// room version does not support restricted join rules
@@ -223,8 +223,9 @@ pub(crate) async fn user_can_perform_restricted_join(
.server_in_room(services.globals.server_name(), &membership.room_id)
.await
{
// Since we can't check this room, mark could_satisfy as false
// so that we can return M_UNABLE_TO_AUTHORIZE_JOIN later.
// Since we can't check this room, mark could_satisfy as
// false so that we can return
// M_UNABLE_TO_AUTHORIZE_JOIN later.
could_satisfy = false;
continue;
}
@@ -252,7 +253,8 @@ pub(crate) async fn user_can_perform_restricted_join(
| Err(_) => Err!(Request(Forbidden("Antispam rejected join request."))),
},
| _ => {
// We don't recognise this join rule, so we cannot satisfy the request.
// We don't recognise this join rule, so we cannot satisfy the
// request.
could_satisfy = false;
debug_info!(
"Unsupported allow rule in restricted join for room {}: {:?}",
@@ -264,16 +266,16 @@ pub(crate) async fn user_can_perform_restricted_join(
}
if could_satisfy {
// We were able to check all the restrictions and can be certain that the
// prospective member is not permitted to join.
// We were able to check all the restrictions and can be certain that
// the prospective member is not permitted to join.
Err!(Request(Forbidden(
"You do not belong to any of the rooms or spaces required to join this room."
)))
} else {
// We were unable to check all the restrictions. This usually means we aren't in
// one of the rooms this one is restricted to, ergo can't check its state for
// the user's membership, and consequently the user *might* be able to join if
// they ask another server.
// We were unable to check all the restrictions. This usually means we
// aren't in one of the rooms this one is restricted to, ergo can't
// check its state for the user's membership, and consequently the
// user *might* be able to join if they ask another server.
Err!(Request(UnableToAuthorizeJoin(
"You do not belong to any of the recognised rooms or spaces required to join this \
room, but this server is unable to verify every requirement. You may be able to \
+6 -6
View File
@@ -173,9 +173,9 @@ async fn process_inbound_transaction(
debug!(pdus = body.pdus.len(), edus = body.edus.len(), "Processing transaction",);
// We have an additional panic catch here because if there's a panic in the
// handle, the sender will effectively become defederated until restart after
// never being removed from the tracking map. Ideally the handle function
// doesn't panic, but when has "ideally" ever worked for us.
// handle, the sender will effectively become defederated until restart
// after never being removed from the tracking map. Ideally the handle
// function doesn't panic, but when has "ideally" ever worked for us.
let handle_result =
std::panic::AssertUnwindSafe(handle(&services, &body.identity, pdus, edus))
.catch_unwind()
@@ -319,9 +319,9 @@ async fn handle_room(
.into_iter()
.map(|(_, event_id, value)| (event_id, value))
.collect();
// Try to sort PDUs by their dependencies, but fall back to arbitrary order on
// failure (e.g., cycles). This is best-effort; proper ordering is the sender's
// responsibility.
// Try to sort PDUs by their dependencies, but fall back to arbitrary order
// on failure (e.g., cycles). This is best-effort; proper ordering is the
// sender's responsibility.
let sorted_event_ids = build_local_dag(&pdu_map, DagBuilderTree::PrevEvents)
.await
.unwrap_or_else(|e| {
+2 -2
View File
@@ -58,8 +58,8 @@ async fn create_join_event(
.acl_check(origin, room_id)
.await?;
// We need to return the state prior to joining, let's keep a reference to that
// here
// We need to return the state prior to joining, let's keep a reference to
// that here
let shortstatehash = services
.rooms
.state
+6 -6
View File
@@ -69,8 +69,8 @@ fn main() {
.write_all(format!("{workspace_packages:?}").as_bytes())
.unwrap();
// built gets the default crate from the workspace. Not sure if this is intended
// behavior, but it's what we want.
// built gets the default crate from the workspace. Not sure if this is
// intended behavior, but it's what we want.
built::write_built_file().expect("Failed to acquire build-time information");
// --- Git Information ---
@@ -133,10 +133,10 @@ fn main() {
// TODO: The git rerun triggers seem to always run
// // Rerun if the git HEAD changes
// println!("cargo:rerun-if-changed=.git/HEAD");
// // Rerun if the ref pointed to by HEAD changes (e.g., new commit on branch)
// if let Some(ref_path) = run_git_command(&["symbolic-ref", "--quiet", "HEAD"])
// { println!("cargo:rerun-if-changed=.git/{ref_path}");
// }
// // Rerun if the ref pointed to by HEAD changes (e.g., new commit on
// branch) if let Some(ref_path) = run_git_command(&["symbolic-ref",
// "--quiet", "HEAD"]) { println!("cargo:rerun-if-changed=.git/{ref_path}"
// ); }
println!("cargo:rerun-if-env-changed=GIT_COMMIT_HASH");
println!("cargo:rerun-if-env-changed=GIT_COMMIT_HASH_SHORT");
+8 -6
View File
@@ -99,8 +99,8 @@ pub fn memory_stats(opts: &str) -> Option<String> {
// Acquire the epoch; ensure latest stats are pulled in
acq_epoch().ok()?;
// SAFETY: calls malloc_stats_print() with our string instance which must remain
// in this frame. https://docs.rs/tikv-jemalloc-sys/latest/tikv_jemalloc_sys/fn.malloc_stats_print.html
// SAFETY: calls malloc_stats_print() with our string instance which must
// remain in this frame. https://docs.rs/tikv-jemalloc-sys/latest/tikv_jemalloc_sys/fn.malloc_stats_print.html
unsafe { ffi::malloc_stats_print(Some(malloc_stats_cb), opaque, opts_p) };
str.truncate(MAX_LENGTH);
@@ -227,7 +227,8 @@ pub mod this_thread {
cell.get_or_init(|| {
let ptr: *const u64 = super::get(&mallctl!(name)).expect("failed to obtain pointer");
// SAFETY: ptr points directly to the internal state of jemalloc for this thread
// SAFETY: ptr points directly to the internal state of jemalloc for
// this thread
unsafe { ptr.as_ref() }.expect("pointer must not be null")
})
}
@@ -291,7 +292,8 @@ pub fn is_phycpu_arena() -> bool { percpu_arenas().is_ok_and(is_equal_to!("phycp
pub fn percpu_arenas() -> Result<&'static str> {
let ptr = get::<*const c_char>(&mallctl!("opt.percpu_arena"))?;
//SAFETY: ptr points to a null-terminated string returned for opt.percpu_arena.
//SAFETY: ptr points to a null-terminated string returned for
// opt.percpu_arena.
let cstr = unsafe { CStr::from_ptr(ptr) };
cstr.to_str().map_err(Into::into)
}
@@ -370,8 +372,8 @@ where
}
fn key(name: &str) -> Result<Key> {
// tikv asserts the output buffer length is tight to the number of required mibs
// so we slice that down here.
// tikv asserts the output buffer length is tight to the number of required
// mibs so we slice that down here.
let segs = name.chars().filter(is_equal_to!(&'.')).count().try_add(1)?;
let name = self::name(name)?;
+2 -1
View File
@@ -122,7 +122,8 @@ pub fn check(config: &Config) -> Result {
));
}
// yeah, unless the user built a debug build hopefully for local testing only
// yeah, unless the user built a debug build hopefully for local testing
// only
if cfg!(not(debug_assertions)) && config.server_name == "your.server.name" {
return Err!(Config(
"server_name",
+19 -17
View File
@@ -47,8 +47,8 @@ impl Manager {
let new = Arc::into_raw(config);
let old = self.active.swap(new.cast_mut(), Ordering::AcqRel);
// SAFETY: The old active pointer was set using an Arc::into_raw(). We're
// obliged to reconstitute that into Arc otherwise it will leak.
// SAFETY: The old active pointer was set using an Arc::into_raw().
// We're obliged to reconstitute that into Arc otherwise it will leak.
Ok(unsafe { Arc::from_raw(old) })
}
@@ -68,17 +68,18 @@ impl Manager {
.as_ref()
.expect("handle was already cached for this thread");
// SAFETY: The caller should not hold multiple references at a time directly
// into Config, as a subsequent reference might invalidate the thread's cache
// causing another reference to dangle.
// SAFETY: The caller should not hold multiple references at a time
// directly into Config, as a subsequent reference might invalidate
// the thread's cache causing another reference to dangle.
//
// This is a highly unusual pattern as most config values are copied by value or
// used immediately without running overlap with another value. Even if it does
// actually occur somewhere, the window of danger is limited to the config being
// reloaded while the reference is held and another access is made by the same
// thread into a different config value. This is mitigated by creating a buffer
// of old configs rather than discarding at the earliest opportunity; the odds
// of this scenario are thus astronomical.
// This is a highly unusual pattern as most config values are copied by
// value or used immediately without running overlap with another
// value. Even if it does actually occur somewhere, the window of
// danger is limited to the config being reloaded while the reference
// is held and another access is made by the same thread into a
// different config value. This is mitigated by creating a buffer
// of old configs rather than discarding at the earliest opportunity;
// the odds of this scenario are thus astronomical.
unsafe { std::mem::transmute(config) }
}
}
@@ -87,8 +88,8 @@ impl Drop for Manager {
fn drop(&mut self) {
let config = self.active.swap(null_mut(), Ordering::AcqRel);
// SAFETY: The active pointer was set using an Arc::into_raw(). We're obliged to
// reconstitute that into Arc otherwise it will leak.
// SAFETY: The active pointer was set using an Arc::into_raw(). We're
// obliged to reconstitute that into Arc otherwise it will leak.
unsafe { Arc::from_raw(config) };
}
}
@@ -119,8 +120,9 @@ fn load_miss(
Arc::from_raw(config)
};
// SAFETY: See the note on the transmute above. The caller should not hold more
// than one reference at a time directly into Config, as the second access
// might invalidate the thread's cache, dangling the reference to the first.
// SAFETY: See the note on the transmute above. The caller should not hold
// more than one reference at a time directly into Config, as the second
// access might invalidate the thread's cache, dangling the reference to
// the first.
unsafe { std::mem::transmute(handle[index].insert(config)) }
}
+2 -1
View File
@@ -2862,7 +2862,8 @@ impl Config {
.extract::<Self>()
.map_err(|e| err!("There was a problem with your configuration file: {e}"))?;
// don't start if we're listening on both UNIX sockets and TCP at same time
// don't start if we're listening on both UNIX sockets and TCP at same
// time
check::is_dual_listening(raw_config)?;
Ok(config)
+3 -2
View File
@@ -221,8 +221,9 @@ impl<E: Event + Clone> TestStore<E> {
) -> Result<Vec<OwnedEventId>> {
let mut auth_chain_sets = vec![];
for ids in event_ids {
// TODO state store `auth_event_ids` returns self in the event ids list
// when an event returns `auth_event_ids` self is not contained
// TODO state store `auth_event_ids` returns self in the event ids
// list when an event returns `auth_event_ids` self is not
// contained
let chain = self
.auth_event_ids(room_id, ids)?
.into_iter()
+63 -53
View File
@@ -195,9 +195,11 @@ where
"auth_check beginning"
);
// [synapse] check that all the events are in the same room as `incoming_event`
// [synapse] check that all the events are in the same room as
// `incoming_event`
// [synapse] do_sig_check check the event has valid signatures for member events
// [synapse] do_sig_check check the event has valid signatures for member
// events
let sender = incoming_event.sender();
@@ -213,7 +215,8 @@ where
return Ok(false);
}
// If the domain of the room_id does not match the domain of the sender, reject
// If the domain of the room_id does not match the domain of the sender,
// reject
if incoming_event.room_id().is_some() {
let Some(room_id_server_name) = incoming_event.room_id().unwrap().server_name()
else {
@@ -230,7 +233,8 @@ where
}
}
// If content.room_version is present and is not a recognized version, reject
// If content.room_version is present and is not a recognized version,
// reject
let content: RoomCreateContentFields = from_json_str(incoming_event.content().get())?;
if content
.room_version
@@ -326,8 +330,8 @@ where
return Ok(false);
}
// If the create event is referenced in the event's auth events, and this is a
// v12 room, reject
// If the create event is referenced in the event's auth events, and this is
// a v12 room, reject
let claims_create_event = incoming_event
.auth_events()
.any(|id| id == room_create_event.event_id());
@@ -335,8 +339,8 @@ where
warn!("event incorrectly references m.room.create event in auth events");
return Ok(false);
} else if !(room_version.room_id_format == RoomIdFormatVersion::V2) && !claims_create_event {
// If the create event is not referenced in the event's auth events, and this is
// a v11 room, reject
// If the create event is not referenced in the event's auth events, and
// this is a v11 room, reject
warn!(
missing = %room_create_event.event_id(),
"event incorrectly did not reference an m.room.create in its auth events"
@@ -545,9 +549,9 @@ where
return Ok(true);
}
// If the event type's required power level is greater than the sender's power
// level, reject If the event has a state_key that starts with an @ and does
// not match the sender, reject.
// If the event type's required power level is greater than the sender's
// power level, reject If the event has a state_key that starts with an @
// and does not match the sender, reject.
if !can_send_event(incoming_event, power_levels_event.as_ref(), sender_power_level) {
warn!(
%sender,
@@ -590,13 +594,13 @@ where
debug!("m.room.power_levels event allowed");
}
// Room version 3: Redaction events are always accepted (provided the event is
// allowed by `events` and `events_default` in the power levels). However,
// servers should not apply or send redaction's to clients until both the
// redaction event and original event have been seen, and are valid. Servers
// should only apply redaction's to events where the sender's domains match, or
// the sender of the redaction has the appropriate permissions per the
// power levels.
// Room version 3: Redaction events are always accepted (provided the event
// is allowed by `events` and `events_default` in the power levels).
// However, servers should not apply or send redaction's to clients until
// both the redaction event and original event have been seen, and are
// valid. Servers should only apply redaction's to events where the
// sender's domains match, or the sender of the redaction has the
// appropriate permissions per the power levels.
if room_version.authorization.special_case_room_redaction
&& *incoming_event.event_type() == TimelineEventType::RoomRedaction
@@ -759,8 +763,8 @@ where
Ok(match target_membership {
| MembershipState::Join => {
trace!("starting target_membership=join check");
// 1. If the only previous event is an m.room.create and the state_key is the
// creator,
// 1. If the only previous event is an m.room.create and the
// state_key is the creator,
// allow
let mut prev_events = current_event.prev_events();
@@ -960,8 +964,8 @@ where
},
| MembershipState::Leave => {
if sender == target_user {
// If the sender matches state_key, allow if that user's current membership
// state is invite, join, or knock.
// If the sender matches state_key, allow if that user's current
// membership state is invite, join, or knock.
let allow = matches!(
target_user_current_membership,
MembershipState::Join | MembershipState::Invite | MembershipState::Knock
@@ -985,10 +989,11 @@ where
);
false
} else {
// If the target user's current membership state is ban, and the sender's power
// level is less than the ban level, reject. If the sender's power level is
// greater than or equal to the kick level, and the target user's power level is
// less than the sender's power level, allow.
// If the target user's current membership state is ban, and the
// sender's power level is less than the ban level, reject.
// If the sender's power level is greater than or equal to
// the kick level, and the target user's power level is less
// than the sender's power level, allow.
let cannot_unban = target_user_current_membership == MembershipState::Ban
&& sender_power < effective_power_levels.ban;
let can_kick =
@@ -1030,9 +1035,9 @@ where
);
false
} else {
// If the sender's power level is greater than or equal to the ban level, and
// the target user's power level is less than the sender's power level, allow.
// Otherwise, reject.
// If the sender's power level is greater than or equal to the
// ban level, and the target user's power level is less than
// the sender's power level, allow. Otherwise, reject.
let allow =
sender_power >= effective_power_levels.ban && target_power < sender_power;
if !allow {
@@ -1047,8 +1052,8 @@ where
allow
},
| MembershipState::Knock if room_version.authorization.knocking => {
// 1. If the `join_rule` is anything other than `knock` or `knock_restricted`,
// reject.
// 1. If the `join_rule` is anything other than `knock` or
// `knock_restricted`, reject.
if !matches!(join_rules, JoinRule::KnockRestricted(_) | JoinRule::Knock) {
warn!(
"Join rule is not set to knock or knock_restricted, knocking is not allowed"
@@ -1057,8 +1062,8 @@ where
} else if matches!(join_rules, JoinRule::KnockRestricted(_))
&& !room_version.authorization.knock_restricted_join_rule
{
// 2. If the `join_rule` is `knock_restricted`, but the room does not support
// `knock_restricted`, reject.
// 2. If the `join_rule` is `knock_restricted`, but the room
// does not support `knock_restricted`, reject.
warn!(
"Join rule is set to knock_restricted but room version does not support \
knock_restricted, knocking is not allowed"
@@ -1076,8 +1081,8 @@ where
sender_membership,
MembershipState::Ban | MembershipState::Invite | MembershipState::Join
) {
// 4. If the `sender`'s current membership is not `ban`, `invite`, or `join`,
// allow.
// 4. If the `sender`'s current membership is not `ban`,
// `invite`, or `join`, allow.
// 5. Otherwise, reject.
warn!(
?target_user_membership_event_id,
@@ -1170,16 +1175,17 @@ fn check_power_levels(
}
// - If any of the keys users_default, events_default, state_default, ban,
// redact, kick, or invite in content are present and not an integer, reject.
// - If either of the keys events or notifications in content are present and
// not a dictionary with values that are integers, reject.
// - If users key in content is not a dictionary with keys that are valid user
// IDs with values that are integers, reject.
// redact, kick, or invite in content are present and not an integer,
// reject.
// - If either of the keys events or notifications in content are present
// and not a dictionary with values that are integers, reject.
// - If users key in content is not a dictionary with keys that are valid
// user IDs with values that are integers, reject.
let user_content: RoomPowerLevelsEventContent =
deserialize_power_levels(power_event.content().get(), room_version)?;
// Validation of users is done in Ruma, synapse for loops validating user_ids
// and integers here
// Validation of users is done in Ruma, synapse for loops validating
// user_ids and integers here
debug!("validation of power event finished");
#[allow(clippy::manual_let_else)]
@@ -1214,8 +1220,8 @@ fn check_power_levels(
let old_state = &current_content;
let new_state = &user_content;
// synapse does not have to split up these checks since we can't combine UserIds
// and EventTypes we do 2 loops
// synapse does not have to split up these checks since we can't combine
// UserIds and EventTypes we do 2 loops
// UserId loop
for user in user_levels_to_check {
@@ -1229,7 +1235,8 @@ fn check_power_levels(
continue;
}
// If the current value is equal to the sender's current power level, reject
// If the current value is equal to the sender's current power level,
// reject
if user != power_event.sender()
&& old_level.map(|i| UserPowerLevel::Int(*i)) == Some(user_level)
{
@@ -1244,8 +1251,9 @@ fn check_power_levels(
return Some(false); // cannot remove ops level == to own
}
// If the current value is higher than the sender's current power level, reject
// If the new value is higher than the sender's current power level, reject
// If the current value is higher than the sender's current power level,
// reject If the new value is higher than the sender's current power
// level, reject
let old_level_too_big = old_level.map(|i| UserPowerLevel::Int(*i)) > Some(user_level);
let new_level_too_big = new_level.map(|i| UserPowerLevel::Int(*i)) > Some(user_level);
if old_level_too_big {
@@ -1280,8 +1288,9 @@ fn check_power_levels(
continue;
}
// If the current value is higher than the sender's current power level, reject
// If the new value is higher than the sender's current power level, reject
// If the current value is higher than the sender's current power level,
// reject If the new value is higher than the sender's current power
// level, reject
let old_level_too_big = old_level.map(|i| UserPowerLevel::Int(*i)) > Some(user_level);
let new_level_too_big = new_level.map(|i| UserPowerLevel::Int(*i)) > Some(user_level);
if old_level_too_big {
@@ -1313,8 +1322,9 @@ fn check_power_levels(
let old_level = old_state.notifications.room;
let new_level = new_state.notifications.room;
if old_level != new_level {
// If the current value is higher than the sender's current power level, reject
// If the new value is higher than the sender's current power level, reject
// If the current value is higher than the sender's current power
// level, reject If the new value is higher than the sender's
// current power level, reject
let old_level_too_big = old_level > user_level;
let new_level_too_big = new_level > user_level;
if old_level_too_big || new_level_too_big {
@@ -1388,8 +1398,8 @@ fn check_redaction(
return Ok(true);
}
// If the domain of the event_id of the event being redacted is the same as the
// domain of the event_id of the m.room.redaction, allow
// If the domain of the event_id of the event being redacted is the same as
// the domain of the event_id of the m.room.redaction, allow
if redaction_event.event_id().server_name()
== redaction_event
.redacts()
+27 -23
View File
@@ -142,8 +142,8 @@ where
// We used to check that all events are events from the correct room
// this is now a check the caller of `resolve` must make.
// Get only the control events with a state_key: "" or ban/kick event (sender !=
// state_key)
// Get only the control events with a state_key: "" or ban/kick event
// (sender != state_key)
let control_events: Vec<_> = all_conflicted
.iter()
.stream()
@@ -187,8 +187,8 @@ where
debug!(count = deduped_power_ev.len(), "deduped power events");
trace!(set = ?deduped_power_ev, "deduped power events");
// This removes the control events that passed auth and more importantly those
// that failed auth
// This removes the control events that passed auth and more importantly
// those that failed auth
let events_to_resolve: Vec<_> = all_conflicted
.iter()
.filter(|&id| !deduped_power_ev.contains(id))
@@ -490,8 +490,8 @@ where
for (node, edges) in graph {
if edges.is_empty() {
let (power_level, origin_server_ts) = key_fn(node.clone()).await?;
// The `Reverse` is because rusts `BinaryHeap` sorts largest -> smallest we need
// smallest -> largest
// The `Reverse` is because rusts `BinaryHeap` sorts largest ->
// smallest we need smallest -> largest
zero_outdegree.push(Reverse(TieBreaker {
power_level,
origin_server_ts,
@@ -507,7 +507,8 @@ where
let mut heap = BinaryHeap::from(zero_outdegree);
// We remove the oldest node (most incoming edges) and check against all other
// We remove the oldest node (most incoming edges) and check against all
// other
let mut sorted = vec![];
// Destructure the `Reverse` and take the smallest `node` each time
while let Some(Reverse(item)) = heap.pop() {
@@ -574,8 +575,8 @@ where
let sender_is_creator =
sender_is_creator(&event, room_version, &auth_events, fetch_event).await?;
// Since v12 the room creators outrank any m.room.power_levels event, so their
// ordering position does not depend on one being in scope.
// Since v12 the room creators outrank any m.room.power_levels event, so
// their ordering position does not depend on one being in scope.
if sender_is_creator
&& room_version
.authorization
@@ -628,8 +629,8 @@ where
.find(|aev| is_type_and_key(*aev, &TimelineEventType::RoomCreate, ""))
.cloned();
// Since v12 the create event is no longer listed in auth_events; the room ID
// is its event ID.
// Since v12 the create event is no longer listed in auth_events; the room
// ID is its event ID.
let create_event = match create_event {
| Some(create_event) => Some(create_event),
| None =>
@@ -738,10 +739,11 @@ where
trace!(map = ?auth_events.keys().collect::<Vec<_>>(), "fetched auth events");
let auth_events = &auth_events;
// NOTE: in state resolution v2.1, auth checks should start with an empty state
// map. It is the caller's job to do this. Previously, this function would
// force an empty state map in this case, and this resulted in power events
// going missing from the resolved state as they'd be discarded here.
// NOTE: in state resolution v2.1, auth checks should start with an empty
// state map. It is the caller's job to do this. Previously, this function
// would force an empty state map in this case, and this resulted in power
// events going missing from the resolved state as they'd be discarded
// here.
let mut resolved_state = unconflicted_state;
for event in events_to_check {
trace!(event_id = event.event_id().as_str(), "checking event");
@@ -774,8 +776,8 @@ where
}
for aid in event.auth_events() {
if let Some(ev) = auth_events.get(aid) {
//TODO: synapse checks "rejected_reason" which is most likely related to
// soft-failing
//TODO: synapse checks "rejected_reason" which is most likely
// related to soft-failing
trace!(event_id = aid.as_str(), "found auth event");
auth_state.insert(
ev.event_type()
@@ -809,8 +811,8 @@ where
debug!(event_id = event.event_id().as_str(), "Running auth checks");
// The key for this is (eventType + a state_key of the signed token not sender)
// so search for it
// The key for this is (eventType + a state_key of the signed token not
// sender) so search for it
let current_third_party = auth_state.iter().find_map(|(_, pdu)| {
(*pdu.event_type() == TimelineEventType::RoomThirdPartyInvite).then_some(pdu)
});
@@ -848,7 +850,8 @@ where
trace!(map = ?resolved_state, "new resolved state");
},
| Ok(false) => {
// synapse passes here on AuthError. We do not add this event to resolved_state.
// synapse passes here on AuthError. We do not add this event to
// resolved_state.
warn!("event {} failed the authentication check", event.event_id());
},
| Err(e) => {
@@ -1276,8 +1279,8 @@ mod tests {
// #[tokio::test]
async fn test_sort() {
for _ in 0..20 {
// since we shuffle the eventIds before we sort them introducing randomness
// seems like we should test this a few times
// since we shuffle the eventIds before we sort them introducing
// randomness seems like we should test this a few times
test_event_sort().await;
}
}
@@ -1917,7 +1920,8 @@ mod tests {
.iter(),
);
// HashMap iteration order is random, so sort this before asserting on it
// HashMap iteration order is random, so sort this before asserting on
// it
for v in conflicted.values_mut() {
v.sort_unstable();
}
+5 -4
View File
@@ -86,8 +86,8 @@ pub(crate) async fn do_check(
// event_id -> StateMap<OwnedEventId>
let mut state_at_event: HashMap<OwnedEventId, StateMap<OwnedEventId>> = HashMap::new();
// Resolve the current state and add it to the state_at_event map then continue
// on in "time"
// Resolve the current state and add it to the state_at_event map then
// continue on in "time"
for node in super::lexicographical_topological_sort(&graph, &|_id| async {
Ok((int!(0), MilliSecondsSinceUnixEpoch(uint!(0))))
})
@@ -172,8 +172,9 @@ pub(crate) async fn do_check(
}
}
// TODO The event is just remade, adding the auth_events and prev_events here
// the `to_pdu_event` was split into `init` and the fn below, could be better
// TODO The event is just remade, adding the auth_events and prev_events
// here the `to_pdu_event` was split into `init` and the fn below,
// could be better
let e = fake_event;
let ev_id = e.event_id();
let event = to_pdu_event(
+2 -2
View File
@@ -45,8 +45,8 @@ impl Module {
.as_ref()
.expect("backing library loaded by this instance");
// SAFETY: Calls dlsym(3) on unix platforms. This might not have to be unsafe
// if wrapped in libloading with_dlerror().
// SAFETY: Calls dlsym(3) on unix platforms. This might not have to be
// unsafe if wrapped in libloading with_dlerror().
let sym = unsafe { handle.get::<Prototype>(cname.as_bytes()) };
let sym = sym.expect("symbol found; binding successful");
+2 -2
View File
@@ -11,8 +11,8 @@ pub fn from_name(name: &str) -> Result<Library> {
}
pub fn from_path(path: &OsStr) -> Result<Library> {
//SAFETY: Calls dlopen(3) on unix platforms. This might not have to be unsafe
// if wrapped in with_dlerror.
//SAFETY: Calls dlopen(3) on unix platforms. This might not have to be
// unsafe if wrapped in with_dlerror.
let lib = unsafe { Library::open(Some(path), OPEN_FLAGS) };
if let Err(e) = lib {
let name = path::to_name(path)?;
+2 -2
View File
@@ -27,8 +27,8 @@ impl fmt::Display for Escape<'_> {
fmt.write_str(&pile_o_bits[last..i])?;
fmt.write_str(s)?;
// NOTE: we only expect single byte characters here - which is fine as long as
// we only match single byte characters
// NOTE: we only expect single byte characters here - which is fine
// as long as we only match single byte characters
last = i.saturating_add(1);
}
+2 -2
View File
@@ -32,8 +32,8 @@ impl<'a> From<&'a str> for &'a Unquoted {
fn from(s: &'a str) -> &'a Unquoted {
let s: &'a str = s.unquote_infallible();
//SAFETY: This is a pattern I lifted from ruma-identifiers for strong-type strs
// by wrapping in a tuple-struct.
//SAFETY: This is a pattern I lifted from ruma-identifiers for
// strong-type strs by wrapping in a tuple-struct.
#[allow(clippy::transmute_ptr_to_ptr)]
unsafe {
std::mem::transmute(s)
+5 -4
View File
@@ -120,10 +120,11 @@ pub fn cores_available() -> impl Iterator<Item = Id> { from_mask(*CORES_AVAILABL
pub fn getcpu() -> Result<usize> {
use crate::{Error, utils::math};
// SAFETY: This is part of an interface with many low-level calls taking many
// raw params, but it's unclear why this specific call is unsafe. Nevertheless
// the value obtained here is semantically unsafe because it can change on the
// instruction boundary trailing its own acquisition and also any other time.
// SAFETY: This is part of an interface with many low-level calls taking
// many raw params, but it's unclear why this specific call is unsafe.
// Nevertheless the value obtained here is semantically unsafe because it
// can change on the instruction boundary trailing its own acquisition and
// also any other time.
let ret: i32 = unsafe { libc::sched_getcpu() };
#[cfg(target_os = "linux")]
+4 -3
View File
@@ -404,7 +404,8 @@ impl<'a, 'de: 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
unhandled!("deserialize Unit not implemented")
}
// this only used for $serde_json::private::RawValue at this time; see MapAccess
// this only used for $serde_json::private::RawValue at this time; see
// MapAccess
#[cfg_attr(unabridged, tracing::instrument(level = "trace", skip_all))]
fn deserialize_identifier<V: Visitor<'de>>(self, visitor: V) -> Result<V::Value> {
let input = "$serde_json::private::RawValue";
@@ -480,8 +481,8 @@ fn deserialize_str(input: &[u8]) -> Result<&str> { string::str_from_bytes(input)
#[cfg(disable)]
#[inline]
fn deserialize_str(input: &[u8]) -> Result<&str> {
// SAFETY: Strings were written by the serializer to the database. Assuming no
// database corruption, the string will be valid. Database corruption is
// SAFETY: Strings were written by the serializer to the database. Assuming
// no database corruption, the string will be valid. Database corruption is
// detected via rocksdb checksums.
unsafe { std::str::from_utf8_unchecked(input) }
}
+2 -2
View File
@@ -109,8 +109,8 @@ pub(crate) fn db_options(config: &Config, env: &Env, row_cache: &Cache) -> Resul
// <https://github.com/facebook/rocksdb/wiki/Track-WAL-in-MANIFEST>
// "We recommend to set track_and_verify_wals_in_manifest to true for
// production, it has been enabled in production for the entire database cluster
// serving the social graph for all Meta apps."
// production, it has been enabled in production for the entire database
// cluster serving the social graph for all Meta apps."
opts.set_track_and_verify_wals_in_manifest(true);
opts.set_paranoid_checks(config.rocksdb_paranoid_file_checks);
+2 -2
View File
@@ -80,8 +80,8 @@ impl super::Map {
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
// Optimization can be `true` if key vector is pre-sorted **by the column
// comparator**.
// Optimization can be `true` if key vector is pre-sorted **by the
// column comparator**.
const SORTED: bool = false;
self.db
+16 -16
View File
@@ -10,25 +10,25 @@ pub(super) fn open(db: &Arc<Engine>, name: &str) -> Arc<ColumnFamily> {
let cf_ptr = bounded_ptr.cast::<ColumnFamily>();
// SAFETY: Column family handles out of RocksDB are basic pointers and can
// be invalidated: 1. when the database closes. 2. when the column is dropped or
// closed. rust_rocksdb wraps this for us by storing handles in their own
// `RwLock<BTreeMap>` map and returning an Arc<BoundColumnFamily<'_>>` to
// provide expected safety. Similarly in "single-threaded mode" we would
// receive `&'_ ColumnFamily`.
// be invalidated: 1. when the database closes. 2. when the column is
// dropped or closed. rust_rocksdb wraps this for us by storing handles in
// their own `RwLock<BTreeMap>` map and returning an
// Arc<BoundColumnFamily<'_>>` to provide expected safety. Similarly in
// "single-threaded mode" we would receive `&'_ ColumnFamily`.
//
// PROBLEM: We need to hold these handles in a field, otherwise we have to take
// a lock and get them by name from this map for every query, which is what
// conduit was doing, but we're not going to make a query for every query so we
// need to be holding it right. The lifetime parameter on these references makes
// that complicated. If this can be done without polluting the userspace
// with lifetimes on every instance of `Map` then this `unsafe` might not be
// necessary.
// PROBLEM: We need to hold these handles in a field, otherwise we have to
// take a lock and get them by name from this map for every query, which is
// what conduit was doing, but we're not going to make a query for every
// query so we need to be holding it right. The lifetime parameter on these
// references makes that complicated. If this can be done without polluting
// the userspace with lifetimes on every instance of `Map` then this
// `unsafe` might not be necessary.
//
// SOLUTION: After investigating the underlying types it appears valid to
// Arc-swap `BoundColumnFamily<'_>` for `ColumnFamily`. They have the
// same inner data, the same Drop behavior, Deref, etc. We're just losing the
// lifetime parameter. We should not hold this handle, even in its Arc, after
// closing the database (dropping `Engine`). Since `Arc<Engine>` is a sibling
// member along with this handle in `Map`, that is prevented.
// same inner data, the same Drop behavior, Deref, etc. We're just losing
// the lifetime parameter. We should not hold this handle, even in its Arc,
// after closing the database (dropping `Engine`). Since `Arc<Engine>` is a
// sibling member along with this handle in `Map`, that is prevented.
unsafe { Arc::from_raw(cf_ptr) }
}
+9 -9
View File
@@ -381,15 +381,15 @@ impl Pool {
// Obtain the result channel.
let chan = cmd.res.take().expect("missing result channel");
// It is worth checking if the future was dropped while the command was queued
// so we can bail without paying for any query.
// It is worth checking if the future was dropped while the command was
// queued so we can bail without paying for any query.
if chan.is_canceled() {
return;
}
// Perform the actual database query. We reuse our database::Map interface but
// limited to the blocking calls, rather than creating another surface directly
// with rocksdb here.
// Perform the actual database query. We reuse our database::Map
// interface but limited to the blocking calls, rather than creating
// another surface directly with rocksdb here.
let result = cmd.map.get_blocking(&cmd.key[0]);
// Send the result back to the submitter.
@@ -416,10 +416,10 @@ impl Drop for Pool {
}
fn into_send_get(result: BatchResult<'_>) -> BatchResult<'static> {
// SAFETY: Necessary to send the Handle (rust_rocksdb::PinnableSlice) through
// the channel. The lifetime on the handle is a device by rust-rocksdb to
// associate a database lifetime with its assets. The Handle must be dropped
// before the database is dropped.
// SAFETY: Necessary to send the Handle (rust_rocksdb::PinnableSlice)
// through the channel. The lifetime on the handle is a device by
// rust-rocksdb to associate a database lifetime with its assets. The
// Handle must be dropped before the database is dropped.
unsafe { std::mem::transmute(result) }
}
+14 -13
View File
@@ -21,11 +21,12 @@ pub(super) fn configure(server: &Arc<Server>) -> (usize, Vec<usize>, Vec<usize>)
let device_name = storage::name_from_path(&path).log_debug_err().ok();
let device_prop = storage::parallelism(&path);
// The default worker count is masked-on if we didn't find better information.
// The default worker count is masked-on if we didn't find better
// information.
let default_worker_count = device_prop.mq.is_empty().then_some(config.db_pool_workers);
// Determine the worker groupings. Each indice represents a hardware queue and
// contains the number of workers which will service it.
// Determine the worker groupings. Each indice represents a hardware queue
// and contains the number of workers which will service it.
let worker_counts: Vec<_> = device_prop
.mq
.iter()
@@ -47,8 +48,8 @@ pub(super) fn configure(server: &Arc<Server>) -> (usize, Vec<usize>, Vec<usize>)
.chain(default_worker_count)
.collect();
// Determine our software queue size for each hardware queue. This is the mpmc
// between the tokio worker and the pool worker.
// Determine our software queue size for each hardware queue. This is the
// mpmc between the tokio worker and the pool worker.
let queue_sizes: Vec<_> = worker_counts
.iter()
.map(|worker_count| {
@@ -58,10 +59,10 @@ pub(super) fn configure(server: &Arc<Server>) -> (usize, Vec<usize>, Vec<usize>)
})
.collect();
// Determine the CPU affinities of each hardware queue. Each indice is a cpu and
// each value is the associated hardware queue. There is a little shiftiness
// going on because cpu's which are not available to the process are filtered
// out, similar to the worker_counts.
// Determine the CPU affinities of each hardware queue. Each indice is a cpu
// and each value is the associated hardware queue. There is a little
// shiftiness going on because cpu's which are not available to the process
// are filtered out, similar to the worker_counts.
let topology = device_prop
.mq
.iter()
@@ -76,8 +77,8 @@ pub(super) fn configure(server: &Arc<Server>) -> (usize, Vec<usize>, Vec<usize>)
topology
});
// Regardless of the capacity of all queues we establish some limit on the total
// number of workers; this is hopefully hinted by nr_requests.
// Regardless of the capacity of all queues we establish some limit on the
// total number of workers; this is hopefully hinted by nr_requests.
let max_workers = device_prop
.mq
.iter()
@@ -92,8 +93,8 @@ pub(super) fn configure(server: &Arc<Server>) -> (usize, Vec<usize>, Vec<usize>)
.sum::<usize>()
.clamp(WORKER_LIMIT.0, max_workers);
// After computing all of the above we can update the global automatic stream
// width, hopefully with a better value tailored to this system.
// After computing all of the above we can update the global automatic
// stream width, hopefully with a better value tailored to this system.
if config.stream_width_scale > 0.0 {
let num_queues = queue_sizes.len().max(1);
update_stream_width(server, num_queues, total_workers);
+15 -13
View File
@@ -133,20 +133,22 @@ fn keyval_longevity<'a, 'b: 'a>(item: KeyVal<'a>) -> KeyVal<'b> {
}
fn slice_longevity<'a, 'b: 'a>(item: &'a Slice) -> &'b Slice {
// SAFETY: The lifetime of the data returned by the rocksdb cursor is only valid
// between each movement of the cursor. It is hereby unsafely extended to match
// the lifetime of the cursor itself. This is due to the limitation of the
// Stream trait where the Item is incapable of conveying a lifetime; this is due
// to GAT's being unstable during its development. This unsafety can be removed
// as soon as this limitation is addressed by an upcoming version.
// SAFETY: The lifetime of the data returned by the rocksdb cursor is only
// valid between each movement of the cursor. It is hereby unsafely
// extended to match the lifetime of the cursor itself. This is due to the
// limitation of the Stream trait where the Item is incapable of conveying
// a lifetime; this is due to GAT's being unstable during its development.
// This unsafety can be removed as soon as this limitation is addressed by
// an upcoming version.
//
// We have done our best to mitigate the implications of this in conjunction
// with the deserialization API such that borrows being held across movements of
// the cursor do not happen accidentally. The compiler will still error when
// values herein produced try to leave a closure passed to a StreamExt API. But
// escapes can happen if you explicitly and intentionally attempt it, and there
// will be no compiler error or warning. This is primarily the case with
// calling collect() without a preceding map(ToOwned::to_owned). A collection
// of references here is illegal, but this will not be enforced by the compiler.
// with the deserialization API such that borrows being held across
// movements of the cursor do not happen accidentally. The compiler will
// still error when values herein produced try to leave a closure passed to
// a StreamExt API. But escapes can happen if you explicitly and
// intentionally attempt it, and there will be no compiler error or
// warning. This is primarily the case with calling collect() without a
// preceding map(ToOwned::to_owned). A collection of references here is
// illegal, but this will not be enforced by the compiler.
unsafe { std::mem::transmute(item) }
}
+2 -1
View File
@@ -15,7 +15,8 @@ macro_rules! unhandled {
// SAFETY: Eliminates branches for serializing and deserializing types never
// encountered in the codebase. This can promote optimization and reduce
// codegen. The developer must verify for every invoking callsite that the
// unhandled type is in no way involved and could not possibly be encountered.
// unhandled type is in no way involved and could not possibly be
// encountered.
unsafe {
std::hint::unreachable_unchecked();
}
+4 -2
View File
@@ -158,7 +158,8 @@ pub(crate) fn update(mut config: Figment, args: &Args) -> Result<Figment> {
config = config.adjoin(("test", &args.test));
// All other individual overrides can go last in case we have options which
// set multiple conf items at once and the user still needs granular overrides.
// set multiple conf items at once and the user still needs granular
// overrides.
for option in &args.option {
let (key, val) = option
.split_once('=')
@@ -172,7 +173,8 @@ pub(crate) fn update(mut config: Figment, args: &Args) -> Result<Figment> {
return Err!("Missing =val in -O/--option: {option:?}");
}
// The value has to pass for what would appear as a line in the TOML file.
// The value has to pass for what would appear as a line in the TOML
// file.
let val = toml::from_str::<FigmentValue>(option)?;
let FigmentValue::Dict(_, val) = val else {
panic!("Unexpected Figment Value: {val:#?}");
+4 -4
View File
@@ -154,8 +154,8 @@ pub(crate) fn init(
let ret = (reload_handles, flame_guard, cap_state);
// Enable the tokio console. This is slightly kludgy because we're judggling
// compile-time and runtime conditions to elide it, each of those changing the
// subscriber's type.
// compile-time and runtime conditions to elide it, each of those changing
// the subscriber's type.
let (console_enabled, console_disabled_reason) = tokio_console_enabled(config);
#[cfg(all(feature = "tokio_console", tokio_unstable))]
if console_enabled {
@@ -169,8 +169,8 @@ pub(crate) fn init(
set_global_default(subscriber);
// If there's a reason the tokio console was disabled when it might be desired
// we output that here after initializing logging
// If there's a reason the tokio console was disabled when it might be
// desired we output that here after initializing logging
if !console_enabled && !console_disabled_reason.is_empty() {
debug_warn!("{console_disabled_reason}");
}
+9 -9
View File
@@ -6,16 +6,16 @@ use conduwuit_core::{debug, info, utils};
#[cold]
pub(super) fn restart() -> ! {
// SAFETY: We have allowed an override for the case where the current_exe() has
// been replaced or removed. By default the server will fail to restart if the
// binary has been replaced (i.e. by cargo); this is for security purposes.
// Command::exec() used to panic in that case.
// SAFETY: We have allowed an override for the case where the current_exe()
// has been replaced or removed. By default the server will fail to restart
// if the binary has been replaced (i.e. by cargo); this is for security
// purposes. Command::exec() used to panic in that case.
//
// We can (and do) prevent that panic by checking the result of current_exe()
// prior to committing to restart, returning an error to the user without any
// unexpected shutdown. In a nutshell that is the excuse for this unsafety.
// Nevertheless, we still want a way to override the restart presentation (i.e.
// admin server restart --force).
// We can (and do) prevent that panic by checking the result of
// current_exe() prior to committing to restart, returning an error to the
// user without any unexpected shutdown. In a nutshell that is the excuse
// for this unsafety. Nevertheless, we still want a way to override the
// restart presentation (i.e. admin server restart --force).
let exe = unsafe { utils::sys::current_exe().expect("program path must be available") };
let envs = env::vars();
let args = env::args().skip(1);
+2 -2
View File
@@ -90,8 +90,8 @@ pub(super) fn shutdown(server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
use conduwuit_core::event;
use tracing::Level;
// The final metrics output is promoted to INFO when tokio_unstable is active in
// a release/bench mode and DEBUG is likely optimized out
// The final metrics output is promoted to INFO when tokio_unstable is
// active in a release/bench mode and DEBUG is likely optimized out
const LEVEL: Level = if cfg!(debug_assertions) {
Level::DEBUG
} else {
+6 -5
View File
@@ -44,8 +44,9 @@ pub(crate) async fn run(services: Arc<Services>) -> Result<()> {
// panics.
services.admin.startup_execute().await?;
// Print first-run banner if necessary. This needs to be done after the startup
// admin commands are run in case one of them created the first user.
// Print first-run banner if necessary. This needs to be done after the
// startup admin commands are run in case one of them created the first
// user.
services.firstrun.print_first_run_banner();
debug!("Running");
@@ -93,9 +94,9 @@ pub(crate) async fn stop(services: Arc<Services>) -> Result<()> {
// unload and explode.
services.stop().await;
// Check that Services and Database will drop as expected, The complex of Arc's
// used for various components can easily lead to references being held
// somewhere improperly; this can hang shutdowns.
// Check that Services and Database will drop as expected, The complex of
// Arc's used for various components can easily lead to references being
// held somewhere improperly; this can hang shutdowns.
debug!("Cleaning up...");
let db = Arc::downgrade(&services.db);
if let Err(services) = Arc::try_unwrap(services) {
+4 -3
View File
@@ -36,7 +36,8 @@ impl super::Service {
// List of commands to execute
let commands = &self.services.server.config.admin_execute;
// Determine if we're running in smoketest-mode which will change some behaviors
// Determine if we're running in smoketest-mode which will change some
// behaviors
let smoketest = self.services.server.config.test.contains("smoke");
// When true, errors are ignored and startup continues.
@@ -55,8 +56,8 @@ impl super::Service {
tokio::task::yield_now().await;
}
// The smoketest functionality is placed here for now and simply initiates
// shutdown after all commands have executed.
// The smoketest functionality is placed here for now and simply
// initiates shutdown after all commands have executed.
if smoketest {
debug_info!("Smoketest mode. All commands complete. Shutting down now...");
self.services
+2 -2
View File
@@ -43,8 +43,8 @@ impl super::Service {
// Use the server user to grant the new admin's power level
let server_user = self.services.globals.server_user.as_ref();
// if this is our local user, just forcefully join them in the room. otherwise,
// invite the remote user.
// if this is our local user, just forcefully join them in the room.
// otherwise, invite the remote user.
if self.services.globals.user_is_local(user_id) {
debug_info!("Inviting local user {user_id} to admin room {room_id}");
self.services
+6 -3
View File
@@ -568,7 +568,8 @@ impl Service {
return None;
}
// Ignore messages from the server user _unless_ the emergency password is set
// Ignore messages from the server user _unless_ the emergency
// password is set
let emergency_password_set = self.services.server.config.emergency_password.is_some();
if event.sender() == server_user && !emergency_password_set {
return None;
@@ -588,7 +589,8 @@ impl Service {
return None;
}
// Only admin users belonging to this server can use escaped commands
// Only admin users belonging to this server can use escaped
// commands
if !self.services.globals.user_is_local(event.sender()) {
return None;
}
@@ -598,7 +600,8 @@ impl Service {
return None;
}
// Escaped commands must be sent locally (via client API), not via federation
// Escaped commands must be sent locally (via client API), not via
// federation
if !sent_locally {
conduwuit::warn!(
"Ignoring escaped admin command from {} that arrived via federation",
+6 -6
View File
@@ -65,8 +65,8 @@ impl crate::Service for Service {
// Process each appservice
for (id, registration) in appservices {
// During startup, resolve any token collisions in favour of appservices
// by logging out conflicting user devices
// During startup, resolve any token collisions in favour of
// appservices by logging out conflicting user devices
if let Some((user_id, device_id, _)) = self
.services
.users
@@ -137,8 +137,8 @@ impl Service {
) -> Result {
//TODO: Check for collisions between exclusive appservice namespaces
// Check for token collision with other appservices (allow re-registration of
// same appservice)
// Check for token collision with other appservices (allow
// re-registration of same appservice)
if let Ok(existing) = self.find_from_token(&registration.as_token).await {
if existing.registration.id != registration.id {
return Err(err!(Request(InvalidParam(
@@ -189,8 +189,8 @@ impl Service {
// remove the appservice from the database
self.db.id_appserviceregistrations.del(appservice_id);
// deletes all active requests for the appservice if there are any so we stop
// sending to the URL
// deletes all active requests for the appservice if there are any so we
// stop sending to the URL
self.services
.sending
.cleanup_events(Some(appservice_id), None, None)
+2 -1
View File
@@ -97,7 +97,8 @@ impl Service {
);
Ok(())
} else {
// logs out any users still in the server service account and removes sessions
// logs out any users still in the server service account and
// removes sessions
self.services.users.deactivate_account(server_user).await
}
}
+11 -9
View File
@@ -186,12 +186,13 @@ impl super::Service {
self.handle_response::<T>(dest, actual, &method, &url, response)
.await,
| Err(error) => {
// This awful wrapping hack is required because `reqwest::Error` does not
// implement `Clone`, but we need to convert it into a local error type to pass
// to should_mark_stale, but handle_error then itself expects the original
// This awful wrapping hack is required because `reqwest::Error`
// does not implement `Clone`, but we need to convert it
// into a local error type to pass to should_mark_stale, but
// handle_error then itself expects the original
// reqwest error.
// handle_error *could* just take the wrapped error, but it'd have to unwrap it
// anyway.
// handle_error *could* just take the wrapped error, but it'd
// have to unwrap it anyway.
let wrapped = Error::Reqwest(error.into());
if self.should_mark_stale(&wrapped) {
debug_info!("{dest} is unhealthy & stale due to a connect error");
@@ -236,8 +237,8 @@ impl super::Service {
const HUGE_ENDPOINTS: [&str; 2] =
["/_matrix/federation/v2/send_join/", "/_matrix/federation/v2/state/"];
let size_limit: u64 = if HUGE_ENDPOINTS.iter().any(|e| url.path().starts_with(e)) {
// Some federation endpoints can return huge response bodies, so we'll bump the
// limit for those endpoints specifically.
// Some federation endpoints can return huge response bodies, so
// we'll bump the limit for those endpoints specifically.
self.services
.server
.config
@@ -260,8 +261,9 @@ impl super::Service {
self.mark_healthy(dest);
})
.map_err(|e| {
// Bad 200 response is usually a sign of bad routing, so we'll mark as stale.
// We won't mark as offline though since we technically got a response.
// Bad 200 response is usually a sign of bad routing, so we'll mark
// as stale. We won't mark as offline though since we technically
// got a response.
self.mark_destination_stale(dest);
err!(BadServerResponse("Server returned bad 200 response: {e:?}"))
})
+4 -4
View File
@@ -121,10 +121,10 @@ impl Service {
if was_unhealthy {
debug!("{} is now healthy", server_name);
}
// The lock for remote_health is deliberately retained until the end of the
// function to prevent parallel requests from marking the server as healthy
// and then immediately marking it as unhealthy again due to the stale cache
// we might be about to clear
// The lock for remote_health is deliberately retained until the end of
// the function to prevent parallel requests from marking the server
// as healthy and then immediately marking it as unhealthy again due
// to the stale cache we might be about to clear
let mut stale_destinations = self.stale_destinations.write();
if stale_destinations.remove(server_name) {
debug!(
+2 -2
View File
@@ -67,8 +67,8 @@ impl crate::Service for Service {
fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
async fn worker(self: Arc<Self>) -> Result {
// first run mode will be enabled if there are no local users, provided it's not
// forcibly disabled for Complement tests
// first run mode will be enabled if there are no local users, provided
// it's not forcibly disabled for Complement tests
let is_first_run = !self.services.config.force_disable_first_run_mode
&& self
.services
+2 -2
View File
@@ -53,8 +53,8 @@ impl Manager {
.spawn(async move { self_.worker().await }),
);
// we can't hold the lock during the iteration with start_worker so the values
// are snapshotted here
// we can't hold the lock during the iteration with start_worker so the
// values are snapshotted here
let services: Vec<Arc<dyn Service>> = self
.service
.read()
+2 -2
View File
@@ -405,8 +405,8 @@ impl Service {
pub fn get_media_file_sha256(&self, key: &[u8]) -> PathBuf {
let mut r = self.get_media_dir();
// Using the hash of the base64 key as the filename
// This is to prevent the total length of the path from exceeding the maximum
// length in most filesystems
// This is to prevent the total length of the path from exceeding the
// maximum length in most filesystems
let digest = <sha2::Sha256 as sha2::Digest>::digest(key);
let encoded = encode_key(&digest);
r.push(encoded);
+2 -1
View File
@@ -311,7 +311,8 @@ impl Service {
let props = html.opengraph.properties;
/* use OpenGraph title/description, but fall back to HTML if not available */
/* use OpenGraph title/description, but fall back to HTML if not
* available */
preview_data.title = props.get("title").cloned().or(html.title);
preview_data.description = props.get("description").cloned().or(html.description);
+3 -2
View File
@@ -427,8 +427,9 @@ impl super::Service {
.moderation
.is_remote_server_media_downloads_forbidden(mxc.server_name)
{
// we'll lie to the client and say the blocked server's media was not found and
// log. the client has no way of telling anyways so this is a security bonus.
// we'll lie to the client and say the blocked server's media was
// not found and log. the client has no way of telling anyways so
// this is a security bonus.
debug_warn!(%mxc, "Received request for media on blocklisted server");
return Err!(Request(NotFound("Media not found.")));
}
+2 -1
View File
@@ -120,7 +120,8 @@ impl super::Service {
.write_to(&mut cursor, image::ImageFormat::Png)
.map_err(|error| err!(error!(%error, "Error writing PNG thumbnail.")))?;
// Save thumbnail in database so we don't have to generate it again next time
// Save thumbnail in database so we don't have to generate it again next
// time
let thumbnail_key = self.db.create_file_metadata(
mxc,
None,
+2 -2
View File
@@ -662,8 +662,8 @@ async fn fix_readreceiptid_readreceipt_duplicates(services: &Services) -> Result
const FIXED_CORRUPT_MSC4133_FIELDS_MARKER: &[u8] = b"fix_corrupt_msc4133_fields";
async fn fix_corrupt_msc4133_fields(services: &Services) -> Result {
// Due to an old bug, some conduwuit databases have `us.cloke.msc4175.tz` user
// profile fields with raw strings instead of quoted JSON ones.
// Due to an old bug, some conduwuit databases have `us.cloke.msc4175.tz`
// user profile fields with raw strings instead of quoted JSON ones.
// This migration fixes that.
use serde_json::{Value, from_slice};
+4 -4
View File
@@ -76,8 +76,8 @@ impl Service {
/// remote from reading our room directory.
#[must_use]
pub fn is_remote_server_room_directory_forbidden(&self, server_name: &ServerName) -> bool {
// Forbidden if NOT (allowed is empty OR allowed contains server OR is self)
// OR forbidden contains server
// Forbidden if NOT (allowed is empty OR allowed contains server OR is
// self) OR forbidden contains server
self.is_remote_server_forbidden(server_name)
|| self
.services
@@ -92,8 +92,8 @@ impl Service {
/// from us.
#[must_use]
pub fn is_remote_server_media_downloads_forbidden(&self, server_name: &ServerName) -> bool {
// Forbidden if NOT (allowed is empty OR allowed contains server OR is self)
// OR forbidden contains server
// Forbidden if NOT (allowed is empty OR allowed contains server OR is
// self) OR forbidden contains server
self.is_remote_server_forbidden(server_name)
|| self
.services
+2 -1
View File
@@ -130,7 +130,8 @@ impl RawScopes {
} else if let Some(captures) = device_token_regex.captures(token) {
scopes.insert(Scope::Device(captures.get(2).unwrap().as_str().into()))
} else if token == "openid" {
// TODO(unspecced): Element sets this scope but doesn't use it for anything
// TODO(unspecced): Element sets this scope but doesn't use
// it for anything
true
} else {
return Err(format!("Invalid scope: {token}"));
+5 -5
View File
@@ -171,9 +171,9 @@ impl crate::Service for Service {
impl Service {
const ACCESS_TOKEN_MAX_AGE: Duration = Duration::from_hours(1);
// Maximum number of pending code grants which will be held in memory at once,
// to prevent unbounded memory use if someone decides to repeatedly reload the
// grant page.
// Maximum number of pending code grants which will be held in memory at
// once, to prevent unbounded memory use if someone decides to repeatedly
// reload the grant page.
const MAX_PENDING_GRANTS: usize = 100;
const RANDOM_TOKEN_LENGTH: usize = 32;
const USER_CODE_CHARACTERS: &[char] = &['0', '1', '2', '3', '4', '5', '6', '7', '8', '9'];
@@ -265,8 +265,8 @@ impl Service {
.host_str()
.is_some_and(|host| ClientMetadata::ACCEPTABLE_LOCALHOSTS.contains(&host))
{
// Remove the port from localhost redirect URIs for native applications when
// checking if it's valid
// Remove the port from localhost redirect URIs for native
// applications when checking if it's valid
stripped_uri.set_port(None).unwrap();
}
+4 -3
View File
@@ -161,8 +161,8 @@ impl Service {
.server_is_ours(room_alias.server_name());
if !server_is_ours {
// TODO: The spec advises servers may cache remote room aliases temporarily.
// We might want to look at doing that.
// TODO: The spec advises servers may cache remote room aliases
// temporarily. We might want to look at doing that.
return self.remote_resolve(room_alias).await;
}
@@ -231,7 +231,8 @@ impl Service {
return Ok(true);
}
// Checking whether the user is able to change canonical aliases of the room
// Checking whether the user is able to change canonical aliases of the
// room
let can_change_canonical_alias = self
.services
.state_accessor
@@ -69,8 +69,9 @@ pub async fn build_local_dag(
let mut id_origin_ts: HashMap<OwnedEventId, _> = HashMap::with_capacity(pdu_map.len());
for (event_id, value) in pdu_map {
// Parse all prev events as event IDs - if they are missing, return an error (we
// can't sanely continue in this case), otherwise skip invalid prev events.
// Parse all prev events as event IDs - if they are missing, return an
// error (we can't sanely continue in this case), otherwise skip
// invalid prev events.
let prev_events = value
.get(tree.as_str())
.and_then(CanonicalJsonValue::as_array)
@@ -94,9 +95,10 @@ pub async fn build_local_dag(
debug!(count = dag.len(), "Sorting incoming events with partial graph");
lexicographical_topological_sort(&dag, &async |node_id| {
// Note: we don't bother fetching power levels because that would massively slow
// this function down. This is a best-effort attempt to order events correctly
// for processing, however ultimately that should be the sender's job.
// Note: we don't bother fetching power levels because that would
// massively slow this function down. This is a best-effort attempt to
// order events correctly for processing, however ultimately that
// should be the sender's job.
let ts = id_origin_ts.get(&node_id).copied().unwrap_or_default();
Ok((int!(0), MilliSecondsSinceUnixEpoch(ts)))
})
@@ -168,14 +170,15 @@ impl super::Service {
assert!(!tail.is_empty(), "empty tail");
assert_ne!(via, self.services.globals.server_name(), "cannot ask ourselves for events");
// The iteration limit is in place to ensure that if the remote server leaves us
// in a state of infinite recursion (as old versions of continuwuity and
// predecessors would), we give up. However, get_missing_events doesn't return
// that many events per-request. Synapse returns 20, and conduwuit+ return 50.
// This means with a hard iteration limit, we might give up too early, before
// we get a chance to even come close to max_fetch_prev_events. As such, we'll
// calculate the limit based on that config option and the aforementioned
// averages.
// The iteration limit is in place to ensure that if the remote server
// leaves us in a state of infinite recursion (as old versions of
// continuwuity and predecessors would), we give up. However,
// get_missing_events doesn't return that many events per-request.
// Synapse returns 20, and conduwuit+ return 50. This means with a
// hard iteration limit, we might give up too early, before
// we get a chance to even come close to max_fetch_prev_events. As such,
// we'll calculate the limit based on that config option and the
// aforementioned averages.
let max_fetch = self.services.server.config.max_fetch_prev_events;
let iteration_limit = max_fetch.saturating_div(20).max(10);
@@ -256,9 +259,9 @@ impl super::Service {
.non_outlier_pdu_exists(&event_id)
.await
{
// NOTE: we explicitly check for *non*-outlier events here, as if we end
// up discovering outlier events, we will be able to upgrade them
// immediately.
// NOTE: we explicitly check for *non*-outlier events here,
// as if we end up discovering outlier events, we will
// be able to upgrade them immediately.
trace!("Already have {event_id} as a timeline PDU");
continue;
}
@@ -286,16 +289,17 @@ impl super::Service {
.non_outlier_pdu_exists(prev_event_id)
.await
{
// NOTE: we explicitly check for *non*-outlier events here, as if we end
// up discovering outlier events, we will be able to upgrade them
// NOTE: we explicitly check for *non*-outlier events
// here, as if we end up discovering outlier
// events, we will be able to upgrade them
// immediately.
trace!("Already have prev event {prev_event_id} as a timeline PDU");
continue;
}
if let Ok(outlier) = self.services.timeline.get_pdu(prev_event_id).await {
// We already have this PDU as an outlier, don't ask for
// it. However, if we are missing any prev events for it, add it to the
// latest events anyway.
// it. However, if we are missing any prev events for
// it, add it to the latest events anyway.
let outlier_missing_prevs = outlier
.prev_events()
.stream()
@@ -333,9 +337,10 @@ impl super::Service {
break;
}
if discovered.len() >= self.services.server.config.max_fetch_prev_events.into() {
// Stupid hack, debug_error!() drops the log to a DEBUG when not in debug mode,
// which is bad because this should at least produce a warning. It's an error in
// debug mode because this can be important, but typically not much can be done
// Stupid hack, debug_error!() drops the log to a DEBUG when not
// in debug mode, which is bad because this should at least
// produce a warning. It's an error in debug mode because
// this can be important, but typically not much can be done
// about it as a user.
#[cfg(debug_assertions)]
error!(elapsed=?start.elapsed(),
@@ -534,8 +539,9 @@ impl super::Service {
todo.push_back(auth_event_id);
have_all_auth = false;
}
// Insert this PDU back at the end of the queue so that it will be resolved once
// all of its auth events have been fetched.
// Insert this PDU back at the end of the queue so that it will
// be resolved once all of its auth events have been
// fetched.
if have_all_auth {
debug!(elapsed=?start.elapsed(),%target_id, "Have all auth events");
discovered_events.insert(target_id, value);
@@ -639,9 +645,10 @@ impl super::Service {
data
},
| Err(e) => {
// NOTE: Unlike fetch_and_handle_auth_events, it's okay if we're missing some
// prevs, since we will fall back to other behaviours for resolving things
// like state. As such, we can just continue as before.
// NOTE: Unlike fetch_and_handle_auth_events, it's okay if
// we're missing some prevs, since we will fall back to
// other behaviours for resolving things like state. As
// such, we can just continue as before.
self.hit_failed_pdu_pull(next_id.clone());
warn!("Failed to fetch prev event {next_id} from any candidate: {e}");
continue;
+14 -12
View File
@@ -103,10 +103,10 @@ impl super::Service {
auth_chain_map.insert(auth_event_id, auth_pdu);
}
// We need to authorise each returned PDU to make sure that the caller can
// correctly detect if one of them is rejected. We don't remove the event from
// the map in case the caller incorrectly then identifies that event as missing,
// rather than rejected.
// We need to authorise each returned PDU to make sure that the caller
// can correctly detect if one of them is rejected. We don't remove
// the event from the map in case the caller incorrectly then
// identifies that event as missing, rather than rejected.
self.authorise_remote_auth_chain(
&auth_chain_map,
room_version_rules,
@@ -142,10 +142,11 @@ impl super::Service {
.collect::<HashMap<_, _>>();
'outer: for (event_id, pdu) in auth_chain_topo {
// If we know the event is rejected OR have it locally, we can skip this check.
// This is safe because if we know the event is rejected, running event auth
// will just reject it again. And, if we have it locally, and HAVEN'T flagged it
// as rejected, then we know it was at least accepted under the check we would
// If we know the event is rejected OR have it locally, we can skip
// this check. This is safe because if we know the event is
// rejected, running event auth will just reject it again. And,
// if we have it locally, and HAVEN'T flagged it as rejected,
// then we know it was at least accepted under the check we would
// otherwise be about to perform, previously.
let (is_rejected, have_locally) = join!(
self.services.pdu_metadata.is_event_rejected(&event_id),
@@ -155,10 +156,11 @@ impl super::Service {
continue;
}
// IMPORTANT: We can't use the handy dandy `handle_outlier_pdu` function here
// because it may then try to fetch missing auth events, resulting in deep
// recursion. We will do the minimum required steps to validate the PDU here.
// Checks 1-3 were already done before this function is called, so we only need
// IMPORTANT: We can't use the handy dandy `handle_outlier_pdu`
// function here because it may then try to fetch missing auth
// events, resulting in deep recursion. We will do the minimum
// required steps to validate the PDU here. Checks 1-3 were
// already done before this function is called, so we only need
// to do check 4.
let mut auth_events_by_key: HashMap<_, _> =
@@ -107,11 +107,11 @@ impl super::Service {
let missing_count = to_fetch.len();
let missing_threshold = max(50, total_count >> 2);
if missing_count >= missing_threshold {
// If there's more than 50 events to fetch, or we're missing 25% or more of the
// state, we would need to make a lot of atomic requests, so we'll just try
// to fetch the full state from the remote instead.
// Since this endpoint might fail in huge rooms, we fall back to atomic fetch
// anyway.
// If there's more than 50 events to fetch, or we're missing 25%
// or more of the state, we would need to make a lot of
// atomic requests, so we'll just try to fetch the full
// state from the remote instead. Since this endpoint might
// fail in huge rooms, we fall back to atomic fetch anyway.
warn!(
elapsed=?start.elapsed(),
%missing_count,
@@ -133,10 +133,10 @@ impl super::Service {
);
let fetched_state = match state_response {
| Ok(Ok(state)) => {
// Filter to ensure we only use the PDUs we were expecting, preventing
// arbitrary state injection.
// Atomic fetch does not have this problem as each PDU is evaluated
// individually.
// Filter to ensure we only use the PDUs we were
// expecting, preventing arbitrary state
// injection. Atomic fetch does not have this
// problem as each PDU is evaluated individually.
let expected: &HashSet<OwnedEventId, BuildHasherDefault<DefaultHasher>> =
&HashSet::from_iter(res.pdu_ids.clone());
state
@@ -48,8 +48,8 @@ impl super::Service {
value: BTreeMap<String, CanonicalJsonValue>,
is_backfilled_event: bool,
) -> Result<Option<RawPduId>> {
// Skip the PDU if we already have it as a timeline event. We still re-process
// outliers in this scenario.
// Skip the PDU if we already have it as a timeline event. We still
// re-process outliers in this scenario.
if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await {
debug!("Database hit for incoming PDU, skipping processing");
return Ok(Some(pdu_id));
@@ -58,9 +58,10 @@ impl super::Service {
"processing incoming PDU from {origin} for room {room_id} with event id {event_id}"
);
// If this is a membership state event for a local user, we will be interested
// in this event even if we aren't in the room. This allows us to process things
// like revoking invites over federation, without being in the room.
// If this is a membership state event for a local user, we will be
// interested in this event even if we aren't in the room. This allows
// us to process things like revoking invites over federation, without
// being in the room.
let is_interesting_member_event = value.get("type").and_then(|t| t.as_str())
== Some("m.room.member")
&& value
@@ -89,14 +90,14 @@ impl super::Service {
)));
}
// If the room doesn't exist (we don't have the create event), there's nothing
// we can do.
// If the room doesn't exist (we don't have the create event), there's
// nothing we can do.
if !room_exists {
return Err!(Request(NotFound("Room is unknown to this server")));
}
// If the room does exist, but we aren't a resident of it, we might be
// interested in an out-of-band membership (for example, an inviter rescinding
// their invite).
// interested in an out-of-band membership (for example, an inviter
// rescinding their invite).
if !is_resident {
if is_interesting_member_event {
// TODO: handle interesting membership events where we aren't in
@@ -127,15 +128,17 @@ impl super::Service {
.handle_outlier_pdu(origin, create_event, event_id, room_id, value)
.await?;
// If this event is being processed as part of backfill, we don't want to end up
// *appending* it during the upgrade process, so we return early.
// If this event is being processed as part of backfill, we don't want
// to end up *appending* it during the upgrade process, so we return
// early.
if is_backfilled_event {
debug!("Not promoting incoming event as it is being backfilled");
return Ok(None);
}
// Skip events sent before we joined (they need to be persisted as backfilled
// events, not timeline events, which is handled elsewhere).
// Skip events sent before we joined (they need to be persisted as
// backfilled events, not timeline events, which is handled
// elsewhere).
let first_ts_in_room = self
.services
.timeline
@@ -150,8 +153,8 @@ impl super::Service {
return Ok(None);
}
// Fetch any missing prev events doing all checks listed here starting at 1.
// These are timeline events.
// Fetch any missing prev events doing all checks listed here starting
// at 1. These are timeline events.
// TODO: This part needs to be done in a background queue somewhere.
debug!("Fetching and persisting any missing prev events");
@@ -30,8 +30,8 @@ impl super::Service {
where
Pdu: Event + Send + Sync,
{
// Skip outlier handling if we already have this event as either a timeline or
// outlier PDU.
// Skip outlier handling if we already have this event as either a
// timeline or outlier PDU.
if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
debug!(
"Database hit for {event_id} (event is either an outlier or already promoted), \
@@ -63,8 +63,8 @@ impl super::Service {
.signature_hash_check_2_3(value, &room_version_rules)
.await?;
// Now that we have checked the signature and hashes we can add the eventID and
// convert to our PduEvent type
// Now that we have checked the signature and hashes we can add the
// eventID and convert to our PduEvent type
incoming_pdu.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
@@ -74,17 +74,18 @@ impl super::Service {
)
.map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?;
// TODO(nex): From hereon the event is not dropped, and thus always added as an
// outlier. However, we only do that at the end of this function, which means
// several duplicated calls to add_pdu_outlier. Shouldn't we just do it here
// instead, since we know it's going to be persisted as an outlier no matter
// what? the rest of this function is basically just to check PDU check 4.
// TODO(nex): From hereon the event is not dropped, and thus always
// added as an outlier. However, we only do that at the end of this
// function, which means several duplicated calls to add_pdu_outlier.
// Shouldn't we just do it here instead, since we know it's going to
// be persisted as an outlier no matter what? the rest of this
// function is basically just to check PDU check 4.
// NOTE^: Technically, persisting the event before knowing if it's rejected
// introduces a race condition in fetch_and_persist_event_auth, where we have
// the event locally, but haven't yet flagged it as rejected, which the
// fetcher perceives as "accepted". I'm not sure if that's practically possible
// though.
// NOTE^: Technically, persisting the event before knowing if it's
// rejected introduces a race condition in
// fetch_and_persist_event_auth, where we have the event locally, but
// haven't yet flagged it as rejected, which the fetcher perceives as
// "accepted". I'm not sure if that's practically possible though.
// Fetch all auth events
let mut auth_events: HashMap<OwnedEventId, PduEvent> = HashMap::new();
@@ -127,7 +128,8 @@ impl super::Service {
}
}
// Ensure none of the auth events are rejected - if they are, reject too.
// Ensure none of the auth events are rejected - if they are, reject
// too.
for (auth_event_id, auth_event) in &auth_events {
if self
.services
@@ -159,8 +161,8 @@ impl super::Service {
}
}
// 4. Reject "due to auth events" if the event doesn't pass auth based on the
// claimed auth events
// 4. Reject "due to auth events" if the event doesn't pass auth based
// on the claimed auth events
debug!("Checking based on auth events");
let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len());
// Build map of auth events
+5 -4
View File
@@ -172,7 +172,8 @@ impl Service {
let next_retry = calc_next_retry(&self.services.server.config, *retries);
if next_retry > last_retry.elapsed() {
// Don't increment the retry if a retry was illegal.
// This can happen if an event was fetched multiple times concurrently.
// This can happen if an event was fetched multiple times
// concurrently.
return;
}
*retries = retries.saturating_add(1);
@@ -184,9 +185,9 @@ impl Service {
/// Removes a PDU from the failed pulls map, allowing it to be re-fetched in
/// future if needed.
pub(super) fn clear_failed_pdu(&self, event_id: &EventId) {
// NOTE: this is a bit pointless since we're unlikely to try pulling an event
// again if we fetch it successfully, but it doesn't hurt to have to prevent the
// map growing indefinitely.
// NOTE: this is a bit pointless since we're unlikely to try pulling an
// event again if we fetch it successfully, but it doesn't hurt to
// have to prevent the map growing indefinitely.
let mut map = self.failed_pdu_pulls.write();
map.remove(event_id);
}
+25 -20
View File
@@ -55,8 +55,8 @@ impl super::Service {
create_event_id: &EventId,
) -> Result<()> {
let event_format = &room_version_rules.event_format;
// NOTE: if we do any more validation outside of deserialisation, it has to be
// done here.
// NOTE: if we do any more validation outside of deserialisation, it has
// to be done here.
if !pdu_fits(pdu_json) {
return Err!(Request(TooLarge("PDU is too large")));
@@ -76,9 +76,10 @@ impl super::Service {
check_event_id_format(auth_event_id, room_version_rules)?;
}
// The m.room.create event is the genesis event and has empty auth_events
// by definition, so it is exempt from the checks below requiring or
// forbidding the create event in auth_events (it cannot reference itself).
// The m.room.create event is the genesis event and has empty
// auth_events by definition, so it is exempt from the checks below
// requiring or forbidding the create event in auth_events (it cannot
// reference itself).
let Some(event_type) = pdu_json.get("type").and_then(CanonicalJsonValue::as_str) else {
return Err!(Request(BadJson("PDU is missing a type")));
};
@@ -190,12 +191,13 @@ impl super::Service {
);
let room_id = incoming_pdu.room_id_or_hash();
// If the incoming event only has one prev event, we can just use the state at
// that event, but otherwise we have to resolve across each fork. If we're
// missing even one of the prev events, we have to ask a remote server for help.
// If the incoming event only has one prev event, we can just use the
// state at that event, but otherwise we have to resolve across each
// fork. If we're missing even one of the prev events, we have to ask
// a remote server for help.
//
// TODO: this can be optimised by only loading auth chain events into memory,
// rather than the entire state.
// TODO: this can be optimised by only loading auth chain events into
// memory, rather than the entire state.
let state_before = self
.state_before_incoming(&incoming_pdu, room_version_rules)
.await?;
@@ -214,10 +216,11 @@ impl super::Service {
if state_before.is_empty()
&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
{
// This can happen if the remote sends an event but cannot be reached to fetch
// the state at it, and all other servers in the room (which might just be the
// unreachable server) are unable to provide required info.
// returning an error here allows the upgrade to be attempted at another time.
// This can happen if the remote sends an event but cannot be
// reached to fetch the state at it, and all other servers in the
// room (which might just be the unreachable server) are unable
// to provide required info. returning an error here allows the
// upgrade to be attempted at another time.
return Err!(Request(Forbidden("Could not resolve incoming state before event")));
}
trace!(state_events = state_before.len(), "Calculated incoming state");
@@ -386,8 +389,9 @@ mod tests {
fn v3_event_id_errors_in_room_v3() {
let v3_event_id = EventId::new_v2_or_v3("zsj67_pqjr5qqh5GMTXqxLM0FqjP5OLrvXO0PjwWe88")
.expect("fixture event hash must be valid");
// Since the urlsafe replacements aren't in the standard base64 alphabet, this
// simply errors instead of decoding to potentially incorrect bytes.
// Since the urlsafe replacements aren't in the standard base64
// alphabet, this simply errors instead of decoding to potentially
// incorrect bytes.
assert!(
check_event_id_format(&v3_event_id, &RoomVersionRules::V3).is_err(),
"V3 event ID should not be valid in room V3"
@@ -397,8 +401,9 @@ mod tests {
fn v3_event_id_ok_in_room_v4_onward() {
let v3_event_id = EventId::new_v2_or_v3("zsj67_pqjr5qqh5GMTXqxLM0FqjP5OLrvXO0PjwWe88")
.expect("fixture event hash must be valid");
// Since the urlsafe replacements aren't in the standard base64 alphabet, this
// simply errors instead of decoding to potentially incorrect bytes.
// Since the urlsafe replacements aren't in the standard base64
// alphabet, this simply errors instead of decoding to potentially
// incorrect bytes.
let expected_bytes = vec![
206, 200, 250, 239, 250, 106, 142, 190, 106, 170, 30, 70, 49, 53, 234, 196, 179, 52,
22, 168, 207, 228, 226, 235, 189, 115, 180, 62, 60, 22, 123, 207,
@@ -409,8 +414,8 @@ mod tests {
expected_bytes,
"V3 event ID in room V4 did not decode to expected bytes"
);
// These versions didn't change the algorithm, but might as well test them
// anyway
// These versions didn't change the algorithm, but might as well test
// them anyway
assert_eq!(
check_event_id_format(&v3_event_id, &RoomVersionRules::V6)
.expect("V3 event should be valid room V6"),
@@ -178,7 +178,8 @@ impl super::Service {
);
return Ok(());
}
// N.B. In a future room version, this will be a soft failure specifically.
// N.B. In a future room version, this will be a soft failure
// specifically.
debug_info!(
via = %ps.via,
"Event is incoming but does not have a valid policy server signature; asking policy \
@@ -400,9 +401,9 @@ impl super::Service {
};
let Some(signatures) = response.signatures.get(via) else {
// NOTE: Legacy policy servers return a `200 {}` to indicate that the event was
// flagged as spam. We'll make a distinction in the error message in case
// it's unexpected.
// NOTE: Legacy policy servers return a `200 {}` to indicate that
// the event was flagged as spam. We'll make a distinction in the
// error message in case it's unexpected.
return Err!(Request(Forbidden("Policy server did not sign the event")));
};
if response.signatures.len() > 1 {
@@ -40,8 +40,8 @@ impl super::Service {
return Err!(Request(Forbidden(debug_info!("Event has been soft-failed"))));
}
// These should never happen, but they're good last-minute sanity checks to
// ensure we never promote totally illegal events.
// These should never happen, but they're good last-minute sanity checks
// to ensure we never promote totally illegal events.
assert_eq!(
*create_event.kind(),
StateEventType::RoomCreate.into(),
@@ -61,10 +61,10 @@ impl super::Service {
let min_depth = self.services.metadata.get_mindepth(room_id).await;
let room_version_rules = get_room_version_rules(create_event)?;
// We now need to resolve the state before the event so that we can perform PDU
// check 5 (event auth passes based on state before the event). To do this, we
// either need to have all the prev events locally, or ask a remote server
// for the state at the event.
// We now need to resolve the state before the event so that we can
// perform PDU check 5 (event auth passes based on state before the
// event). To do this, we either need to have all the prev events
// locally, or ask a remote server for the state at the event.
let (passes_state_before, state_before) = self
.state_before_check_5(&incoming_pdu, &room_version_rules, create_event, origin)
.await?;
@@ -77,13 +77,13 @@ impl super::Service {
}
// Now that we know the event passes both self-authentication, and
// authentication based on the state before the event, we need to check that it
// passes based on the *current* room state (state across all forward
// extremities). If it doesn't, we accept it, but soft-fail it, and this
// prevents it being promoted.
// authentication based on the state before the event, we need to check
// that it passes based on the *current* room state (state across all
// forward extremities). If it doesn't, we accept it, but soft-fail
// it, and this prevents it being promoted.
// We lock the room here to prevent the current state from changing beneath us
// mid-check.
// We lock the room here to prevent the current state from changing
// beneath us mid-check.
trace!(
room_id = %room_id,
"Locking the room"
@@ -102,8 +102,8 @@ impl super::Service {
})?;
// Determine whether this PDU should be soft-failed.
// If the auth check failed, invariably yes. Otherwise, only if the user isn't
// allowed to redact the target event (if any).
// If the auth check failed, invariably yes. Otherwise, only if the user
// isn't allowed to redact the target event (if any).
let redaction_permitted =
if let Some(redacted_id) = incoming_pdu.redacts_id(&room_version_rules) {
self.services
@@ -117,10 +117,11 @@ impl super::Service {
let mut should_soft_fail = !redaction_permitted || !passes_current_state;
if !should_soft_fail {
// Now we can perform check 7, which is ensuring the event passes policy server
// checks.
// We explicitly only do this if we aren't already going to soft-fail the event,
// since the policy server refusing this event also soft-fails it.
// Now we can perform check 7, which is ensuring the event passes
// policy server checks.
// We explicitly only do this if we aren't already going to
// soft-fail the event, since the policy server refusing this
// event also soft-fails it.
debug!("Checking policy server for event");
should_soft_fail = !self
.policy_server_check_7(&incoming_pdu, &mut val, &room_version_rules)
@@ -133,9 +134,10 @@ impl super::Service {
}
})?;
// TODO: this is supposed to hide redactions from policy servers and janitorial
// bots, however, for full efficacy it also needs to hide redactions for
// unknown events. This needs to be investigated at a later time.
// TODO: this is supposed to hide redactions from policy servers and
// janitorial bots, however, for full efficacy it also needs to
// hide redactions for unknown events. This needs to be
// investigated at a later time.
if let Some(redact_id) = incoming_pdu.redacts_id(&room_version_rules) {
debug!(
redact_id = %redact_id,
@@ -161,8 +163,8 @@ impl super::Service {
);
}
// The PDU has now passed all checks! We can now promote it (or soft-fail it if
// the verdict is such).
// The PDU has now passed all checks! We can now promote it (or
// soft-fail it if the verdict is such).
trace!("Appending pdu to timeline");
let pdu_id = self
.services
+38 -32
View File
@@ -226,16 +226,17 @@ impl Service {
&& matches!(join_rules, JoinRule::Restricted(_) | JoinRule::KnockRestricted(_))
{
if room_version_rules.authorization.restricted_join_rule {
// This is a restricted room, check if we can complete the join requirements
// locally.
// This is a restricted room, check if we can complete the join
// requirements locally.
let needs_auth_user = self
.user_can_perform_restricted_join(sender_user, room_id)
.await;
if needs_auth_user.is_ok_and(is_true!()) {
// If there was an error or the value is false, we'll try joining over
// federation. Since it's Ok(true), we can authorise this locally.
// If we can't select a local user, this will remain None, the join will fail,
// and we'll fall back to federation.
// If there was an error or the value is false, we'll try
// joining over federation. Since it's Ok(true), we can
// authorise this locally. If we can't select a local
// user, this will remain None, the join will fail, and
// we'll fall back to federation.
auth_user = self
.select_authorising_user(room_id, sender_user, &state_lock)
.await
@@ -370,8 +371,8 @@ impl Service {
// Remove event id if it exists
join_event_stub.remove("event_id");
// In order to create a compatible ref hash (EventID) the `hashes` field needs
// to be present
// In order to create a compatible ref hash (EventID) the `hashes` field
// needs to be present
self.services
.server_keys
.hash_and_sign_event(&mut join_event_stub, &room_version_rules)?;
@@ -394,11 +395,11 @@ impl Service {
.await,
);
// NOTE: send_join can take a long time to respond, but from the point of view
// of other servers, we may already have finished joining. This means they
// sometimes end up sending PDUs to us that we aren't yet ready to accept, and
// consequently drop. Holding the mutex over the room while processing mitigates
// this.
// NOTE: send_join can take a long time to respond, but from the point
// of view of other servers, we may already have finished joining.
// This means they sometimes end up sending PDUs to us that we aren't
// yet ready to accept, and consequently drop. Holding the mutex over
// the room while processing mitigates this.
let _room_lock = self
.services
.event_handler
@@ -628,9 +629,9 @@ impl Service {
debug!("Updating joined counts for new room");
self.services.state_cache.update_joined_count(room_id).await;
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
// We append to state before appending the pdu, so we don't have a
// moment in time with the pdu without it's state. This is okay
// because append_pdu can't fail.
let statehash_after_join = self
.services
.state
@@ -654,17 +655,18 @@ impl Service {
.await;
info!("Setting final room state for new room");
// We set the room state after inserting the pdu, so that we never have a moment
// in time where events in the current room state do not exist
// We set the room state after inserting the pdu, so that we never have
// a moment in time where events in the current room state do not
// exist
self.services
.state
.set_room_state(room_id, statehash_after_join, &state_lock);
if !resident_before {
// NOTE: We replace local extremities for this room if we were not a resident
// before. We might be doing a remote join to satisfy restricted join rules,
// so we don't want to do this if we're already a resident. Otherwise, we
// want to replace our forward extremities whole-sale in case we were
// desynced.
// NOTE: We replace local extremities for this room if we were not a
// resident before. We might be doing a remote join to satisfy
// restricted join rules, so we don't want to do this if we're
// already a resident. Otherwise, we want to replace our forward
// extremities whole-sale in case we were desynced.
info!("Replacing local forward extremities");
self.services
.state
@@ -821,7 +823,8 @@ impl Service {
)
.await
else {
// No join rules means there's nothing to authorise (defaults to invite)
// No join rules means there's nothing to authorise (defaults to
// invite)
return Ok(false);
};
@@ -847,8 +850,9 @@ impl Service {
.server_in_room(self.services.globals.server_name(), &membership.room_id)
.await
{
// Since we can't check this room, mark could_satisfy as false
// so that we can return M_UNABLE_TO_AUTHORIZE_JOIN later.
// Since we can't check this room, mark could_satisfy as
// false so that we can return
// M_UNABLE_TO_AUTHORIZE_JOIN later.
could_satisfy = false;
continue;
}
@@ -877,7 +881,8 @@ impl Service {
| Err(_) => Err!(Request(Forbidden("Antispam rejected join request."))),
},
| _ => {
// We don't recognise this join rule, so we cannot satisfy the request.
// We don't recognise this join rule, so we cannot satisfy
// the request.
could_satisfy = false;
debug_info!(
"Unsupported allow rule in restricted join for room {}: {:?}",
@@ -889,15 +894,16 @@ impl Service {
}
if could_satisfy {
// We were able to check all the restrictions and can be certain that the
// prospective member is not permitted to join.
// We were able to check all the restrictions and can be certain
// that the prospective member is not permitted to join.
Err!(Request(Forbidden(
"You do not belong to any of the rooms or spaces required to join this room."
)))
} else {
// We were unable to check all the restrictions. This usually means we aren't in
// one of the rooms this one is restricted to, ergo can't check its state for
// the user's membership, and consequently the user *might* be able to join if
// We were unable to check all the restrictions. This usually means
// we aren't in one of the rooms this one is restricted to, ergo
// can't check its state for the user's membership, and
// consequently the user *might* be able to join if
// they ask another server.
Err!(Request(UnableToAuthorizeJoin(
"You do not belong to any of the recognised rooms or spaces required to join \
@@ -78,8 +78,8 @@ impl super::Service {
let mut bundled = BundledMessageLikeRelations::<Box<serde_json::value::RawValue>>::new();
// Handle m.replace relations - find the most recent valid one (lazy load
// original event)
// Handle m.replace relations - find the most recent valid one (lazy
// load original event)
if !replace_events.is_empty() {
if let Some(replacement) = self
.find_most_recent_valid_replacement(user_id, pdu, &replace_events)
@@ -139,7 +139,8 @@ impl super::Service {
let next = match result {
| None => Some(pdu),
| Some(current) => {
// Compare by origin_server_ts first, then event_id lexicographically
// Compare by origin_server_ts first, then event_id
// lexicographically
match pdu.origin_server_ts().cmp(&current.origin_server_ts()) {
| std::cmp::Ordering::Greater => Some(pdu),
| std::cmp::Ordering::Equal if pdu.event_id() > current.event_id() =>
+2 -2
View File
@@ -63,8 +63,8 @@ impl Data {
from: PduCount,
dir: Direction,
) -> impl Stream<Item = PdusIterItem> + Send + 'a {
// Query from exact position then filter excludes it (saturating_inc could skip
// events at min/max boundaries)
// Query from exact position then filter excludes it (saturating_inc
// could skip events at min/max boundaries)
let from_unsigned = from.into_unsigned();
let mut current = ArrayVec::<u8, 16>::new();
current.extend(target.to_be_bytes());
+2 -2
View File
@@ -163,8 +163,8 @@ impl Service {
&'a self,
query: &'a RoomQuery<'_>,
) -> Result<impl Stream<Item = RawPduId> + Send + 'a + use<'a>> {
// result is modeled as a stream such that callers don't have to be refactored
// though an additional async/wrap still exists for now
// result is modeled as a stream such that callers don't have to be
// refactored though an additional async/wrap still exists for now
let shortroomid = self.services.short.get_shortroomid(query.room_id).await?;
let pdu_ids = self.search_pdu_ids_query_room(query, shortroomid).await;
@@ -43,7 +43,8 @@ impl super::Service {
match history_visibility {
| HistoryVisibility::Invited => {
// Allow if any member on requesting server was AT LEAST invited, else deny
// Allow if any member on requesting server was AT LEAST
// invited, else deny
current_server_members
.any(async |member| self.user_was_invited(shortstatehash, &member).await)
.await
+2 -1
View File
@@ -92,7 +92,8 @@ impl super::Service {
match history_visibility {
| HistoryVisibility::Invited => {
// Allow if any member on requesting server was AT LEAST invited, else deny
// Allow if any member on requesting server was AT LEAST
// invited, else deny
self.user_was_invited(shortstatehash, user_id).await
},
| HistoryVisibility::Joined => {
+3 -3
View File
@@ -327,9 +327,9 @@ impl super::Service {
last_state: Vec<RawStrippedState>,
invite_via: Option<Vec<OwnedServerName>>,
) -> Result<()> {
// return an error for blocked invites. ignored invites aren't handled here
// since the recipient's membership should still be changed to `invite`.
// they're filtered out in the individual /sync handlers
// return an error for blocked invites. ignored invites aren't handled
// here since the recipient's membership should still be changed to
// `invite`. they're filtered out in the individual /sync handlers
if matches!(
self.services
.users
+29 -26
View File
@@ -129,9 +129,9 @@ impl Service {
// We have this room locally.
let children_state = self.get_space_child_events(room_id, suggested_only).await;
// All of the room's children are accessible to this server (because we have the
// full room and its state), although some of them may not be accessible to
// the querying user.
// All of the room's children are accessible to this server
// (because we have the full room and its state), although
// some of them may not be accessible to the querying user.
(SpaceHierarchyRoomsChunk::new(summary, children_state), vec![])
} else if let Some(via) = via
&& let Some((
@@ -197,11 +197,11 @@ impl Service {
max_depth: Option<UInt>,
suggested_only: bool,
) -> Result<Accessibility<Vec<SpaceHierarchyRoomsChunk>>> {
// This function traverses the space hierarchy tree depth-first as required by
// the specification.
// This function traverses the space hierarchy tree depth-first as
// required by the specification.
// Check accessibility of the root room first, because we need to error out
// if it isn't accessible.
// Check accessibility of the root room first, because we need to error
// out if it isn't accessible.
// TODO refactor this once the Try trait is stable
let root_summary = match self
.get_room_summary_and_children_for_user(
@@ -222,27 +222,29 @@ impl Service {
let mut summaries = vec![root_summary.summary];
let mut inaccessible_children: HashSet<_> =
root_summary.inaccessible_children.into_iter().collect();
// Rooms already included in the response. The spec requires each room to
// appear at most once, and this also terminates traversal of cyclic space
// graphs (e.g. a space containing itself).
// Rooms already included in the response. The spec requires each room
// to appear at most once, and this also terminates traversal of
// cyclic space graphs (e.g. a space containing itself).
let mut visited: HashSet<OwnedRoomId> = HashSet::from([room_id]);
// TODO refactor this with Vec::peek_mut once it's stabilized
while let Some(layer) = queue.last_mut() {
let Some(SpaceChild { room_id, via }) = layer.pop() else {
// If this layer is empty, discard it from the queue and continue
// If this layer is empty, discard it from the queue and
// continue
queue.pop();
continue;
};
// The root room is at depth zero, so the current layer is the depth of
// the child about to be processed.
// The root room is at depth zero, so the current layer is the depth
// of the child about to be processed.
let depth = u64::try_from(queue.len()).expect("hierarchy depth should fit into u64");
if max_depth.is_some_and(|max_depth| depth > max_depth.into()) {
continue;
}
// Do not request rooms which have been determined to be inaccessible
// Do not request rooms which have been determined to be
// inaccessible
if inaccessible_children.contains(&room_id) {
continue;
}
@@ -280,11 +282,11 @@ impl Service {
summaries.push(summary.summary);
inaccessible_children.extend(summary.inaccessible_children);
// Bail out if queue length exceeds 50 to keep cycles in the space graph from
// blowing up the traversal. If you actually have over fifty nested spaces,
// and you're looking at this function to figure out what the issue is, I
// suggest you reconsider some of the choices you made which led you to this
// point.
// Bail out if queue length exceeds 50 to keep cycles in the space
// graph from blowing up the traversal. If you actually have over
// fifty nested spaces, and you're looking at this function to
// figure out what the issue is, I suggest you reconsider some of
// the choices you made which led you to this point.
if queue.len() > 50 {
return Err!(Request(TooLarge("Space hierarchy is unreasonably large")));
}
@@ -601,9 +603,9 @@ impl Service {
return false;
}
// Servers may always see summaries if any of their users are participating in
// the room. It's the requesting server's job to restrict visibility on a
// per-user basis.
// Servers may always see summaries if any of their users are
// participating in the room. It's the requesting server's job to
// restrict visibility on a per-user basis.
if self
.services
.state_cache
@@ -613,8 +615,9 @@ impl Service {
return true;
}
// For restricted rooms, federation responses must include the summary when
// a user on the querying server can satisfy one of the join conditions.
// For restricted rooms, federation responses must include the summary
// when a user on the querying server can satisfy one of the join
// conditions.
if let JoinRuleSummary::Restricted(RestrictedSummary { allowed_room_ids, .. }) =
&summary.join_rule
{
@@ -630,8 +633,8 @@ impl Service {
}
}
// If the server isn't in the room, the same visibility rules apply as for
// anonymous summary requests.
// If the server isn't in the room, the same visibility rules apply as
// for anonymous summary requests.
self.user_may_see_summary(None, summary).await
}
}
+36 -30
View File
@@ -72,9 +72,9 @@ impl super::Service {
.map(Arc::new)
.await;
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
// We append to state before appending the pdu, so we don't have a
// moment in time with the pdu without it's state. This is okay
// because append_pdu can't fail.
self.services
.state
.set_event_state(&pdu.event_id, &room_id, state_ids_compressed)
@@ -91,8 +91,9 @@ impl super::Service {
// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
// are not added as forward extremities.
// This also means we set the state here.
// We do this BEFORE setting the extremities so that there's never a point in
// time where we have fresh extremities referencing stale state.
// We do this BEFORE setting the extremities so that there's never a
// point in time where we have fresh extremities referencing stale
// state.
let mut extremities: Vec<_> = self
.services
.state
@@ -177,12 +178,14 @@ impl super::Service {
let start = Instant::now();
trace!("Calculating extremities");
// Start with the current extremity set, to avoid dropping unreferenced events.
// Start with the current extremity set, to avoid dropping unreferenced
// events.
let mut new_extremities: HashSet<OwnedEventId> =
current_extremities.into_iter().collect();
// Add the incoming event
new_extremities.insert(incoming_pdu.event_id().to_owned());
// Remove any extremities that have since been referenced but not removed(?)
// Remove any extremities that have since been referenced but not
// removed(?)
new_extremities = new_extremities
.into_iter()
.stream()
@@ -197,16 +200,17 @@ impl super::Service {
.collect()
.await;
// Also iteratively walk prev events to ensure we drop any ancestors which are
// referenced regardless of soft-fail or rejection status. Soft-failed events
// and rejected events are not added as forward extremities, and as such do not
// remove their own prevs from the extremity set. This can lead to a situation
// where extremities build up because there's lots of soft-fails in the room,
// ultimately resulting in excessive cross-state boundary dummy events being
// Also iteratively walk prev events to ensure we drop any ancestors
// which are referenced regardless of soft-fail or rejection status.
// Soft-failed events and rejected events are not added as forward
// extremities, and as such do not remove their own prevs from the
// extremity set. This can lead to a situation where extremities build
// up because there's lots of soft-fails in the room, ultimately
// resulting in excessive cross-state boundary dummy events being
// created to squash the references. By walking the prevs until we stop
// reaching soft-fails, we can ensure that we properly prune all events which
// have actually been referenced, regardless of their soft-fail or rejection
// status.
// reaching soft-fails, we can ensure that we properly prune all events
// which have actually been referenced, regardless of their soft-fail
// or rejection status.
let mut ancestors_todo: VecDeque<OwnedEventId> =
VecDeque::from_iter(incoming_pdu.prev_events.clone());
let room_id = incoming_pdu.room_id_or_hash();
@@ -376,14 +380,15 @@ impl super::Service {
.await
.map_err(|_| err!(Database("Room does not exist")))?;
// Make unsigned fields correct. This is not properly documented in the spec,
// but state events need to have previous content in the unsigned field, so
// clients can easily interpret things like membership changes
// Make unsigned fields correct. This is not properly documented in the
// spec, but state events need to have previous content in the
// unsigned field, so clients can easily interpret things like
// membership changes
// TODO: This needs to be refactored to add this information on the fly.
// Because the prev content becomes part of the unsigned object in the PDU, we
// unintentionally leak redacted or hidden content to local users.
// See: https://forgejo.ellis.link/continuwuation/continuwuity/issues/1103
// Because the prev content becomes part of the unsigned object in the
// PDU, we unintentionally leak redacted or hidden content to local
// users. See: https://forgejo.ellis.link/continuwuation/continuwuity/issues/1103
self.populate_unsigned(pdu, &mut pdu_json).await;
// We must keep track of all events that have been referenced.
@@ -600,13 +605,13 @@ 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
// 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
// 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
self.services
.state_cache
.update_membership(room_id, &target_user_id, pdu, true)
@@ -706,9 +711,10 @@ impl super::Service {
/// 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
// their relation / thread. We need some kind of way to trigger when we receive
// this event, and potentially a way to rebuild the table entirely.
// CONCERN: If we receive events with a relation out-of-order, we never
// write their relation / thread. We need some kind of way to trigger
// when we receive this event, and potentially a way to rebuild the
// table entirely.
if let Ok(content) = pdu.get_content::<ExtractRelatesToEventId>() {
if let Ok(related_pducount) = self.get_pdu_count(&content.relates_to.event_id).await {
+15 -13
View File
@@ -41,7 +41,8 @@ impl super::Service {
self.check_pdu_for_admin_room(&pdu, sender).boxed().await?;
}
// If redaction event is not authorized, do not append it to the timeline
// If redaction event is not authorized, do not append it to the
// timeline
if *pdu.kind() == TimelineEventType::RoomRedaction {
use ruma::RoomVersionId::*;
trace!("Running redaction checks for room {room_id}");
@@ -105,16 +106,16 @@ impl super::Service {
.await;
}
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
// We append to state before appending the pdu, so we don't have a
// moment in time with the pdu without it's state. This is okay
// because append_pdu can't fail.
trace!("Appending {} state for room {room_id}", pdu.event_id());
let statehashid = self.services.state.append_to_state(&pdu, &room_id).await?;
trace!("State hash ID for {room_id}: {statehashid:?}");
// prev_events is capped at the room version's limit, so this PDU does not
// necessarily reference every extremity. Any it did not reference is still a
// head of the DAG and has to be kept.
// prev_events is capped at the room version's limit, so this PDU does
// not necessarily reference every extremity. Any it did not reference
// is still a head of the DAG and has to be kept.
let prev_events: HashSet<&EventId> = pdu.prev_events().collect();
let unreferenced: Vec<OwnedEventId> = self
.services
@@ -159,8 +160,9 @@ impl super::Service {
}
}
// We set the room state after inserting the pdu, so that we never have a moment
// in time where events in the current room state do not exist
// We set the room state after inserting the pdu, so that we never have
// a moment in time where events in the current room state do not
// exist
trace!("Setting room state for room {room_id}");
self.services
.state
@@ -187,8 +189,8 @@ impl super::Service {
.collect()
.await;
// In case we are kicking or banning a user, we need to inform their server of
// the change
// In case we are kicking or banning a user, we need to inform their
// server of the change
if *pdu.kind() == TimelineEventType::RoomMember {
if let Some(state_key_uid) = &pdu
.state_key
@@ -199,8 +201,8 @@ impl super::Service {
}
}
// Remove our server from the server list since it will be added to it by
// room_servers() and/or the if statement above
// Remove our server from the server list since it will be added to it
// by room_servers() and/or the if statement above
servers.remove(self.services.globals.server_name());
trace!("Sending PDU {} to {} servers", pdu.event_id(), servers.len());
+7 -5
View File
@@ -25,7 +25,8 @@ use super::RoomMutexGuard;
/// Ensures the given PDU fits inside the size limits for a PDU.
#[must_use]
pub fn pdu_fits(owned_obj: &CanonicalJsonObject) -> bool {
// room IDs, event IDs, senders, types, and state keys must all be <= 255 bytes
// room IDs, event IDs, senders, types, and state keys must all be <= 255
// bytes
if let Some(CanonicalJsonValue::String(room_id)) = owned_obj.get("room_id") {
if room_id.len() > 255 {
return false;
@@ -341,10 +342,11 @@ impl super::Service {
"Checking event in room {} with policy server",
pdu.room_id.as_ref().map_or("None", |id| id.as_str())
);
// We need to remove the event ID before getting a PS signature on the event.
// Note that we seemingly pointlessly add it above just to remove it here, but
// it's important to make sure the event ID isn't the field that makes the
// difference between an illegally-large event and one that is okay.
// We need to remove the event ID before getting a PS signature on
// the event. Note that we seemingly pointlessly add it above
// just to remove it here, but it's important to make sure the
// event ID isn't the field that makes the difference between an
// illegally-large event and one that is okay.
pdu_json.remove("event_id");
self.services
.event_handler
+4 -3
View File
@@ -113,8 +113,8 @@ impl super::Service {
room_id: &RoomId,
json: &Raw<AnyStateEventContent>,
) -> Result {
// prevents common ACL paw-guns as ACL management is difficult and prone to
// irreversible mistakes
// prevents common ACL paw-guns as ACL management is difficult and prone
// to irreversible mistakes
let acl_content = json
.deserialize_as_unchecked::<RoomServerAclEventContent>()
@@ -203,7 +203,8 @@ impl super::Service {
room_id: &RoomId,
json: &Raw<AnyStateEventContent>,
) -> Result {
// admin room is a sensitive room, it should not ever be made world readable
// admin room is a sensitive room, it should not ever be made world
// readable
if let Ok(admin_room_id) = self.services.admin.get_admin_room().await
&& admin_room_id == room_id
+6 -5
View File
@@ -545,8 +545,8 @@ impl Service {
continue;
}
// Empty prev id forces synapse to resync; because synapse resyncs,
// we can just insert placeholder data
// Empty prev id forces synapse to resync; because synapse
// resyncs, we can just insert placeholder data
let edu = Edu::DeviceListUpdate(DeviceListUpdateContent::new(
user_id,
device_id!("placeholder").to_owned(),
@@ -876,7 +876,8 @@ impl Service {
}
for pdu in pdus {
// Redacted events are not notification targets (we don't send push for them)
// Redacted events are not notification targets (we don't send push
// for them)
if pdu.is_redacted() {
continue;
}
@@ -1011,8 +1012,8 @@ impl Service {
// TODO: remove all unsigned data over federation.
unsigned.remove("transaction_id");
}
// We don't support any room versions that have the event_id in the PDU JSON,
// so we can safely remove it here.
// We don't support any room versions that have the event_id in the PDU
// JSON, so we can safely remove it here.
pdu_json.remove("event_id");
std::future::ready(
+2 -1
View File
@@ -24,7 +24,8 @@ fn load(db: &Arc<Database>) -> Result<Box<Ed25519KeyPair>> {
let (version, key) = db["global"]
.get_blocking(b"keypair")
.map(|ref val| {
// database deserializer is having trouble with this so it's manual for now
// database deserializer is having trouble with this so it's manual
// for now
let mut elems = val.split(|&b| b == b'\xFF');
let vlen = elems.next().expect("invalid keypair entry").len();
let ver = string_from_bytes(&val[..vlen]).expect("invalid keypair version");
+2 -2
View File
@@ -146,8 +146,8 @@ impl Services {
.start()
.await?;
// reset dormant online/away statuses to offline, and set the server user as
// online
// reset dormant online/away statuses to offline, and set the server
// user as online
if self.server.config.allow_local_presence {
self.presence.unset_all_presence().await;
_ = self
+8 -6
View File
@@ -24,8 +24,9 @@ pub fn create(services: Arc<Services>) -> (State, Guard) {
impl Drop for Guard {
fn drop(&mut self) {
let ptr = Arc::as_ptr(&self.services);
// SAFETY: Parity with Arc::into_raw() called in create(). This revivifies the
// Arc lost to State so it can be dropped, otherwise Services will leak.
// SAFETY: Parity with Arc::into_raw() called in create(). This
// revivifies the Arc lost to State so it can be dropped, otherwise
// Services will leak.
let arc = unsafe { Arc::from_raw(ptr) };
debug_assert!(
Arc::strong_count(&arc) > 1,
@@ -54,10 +55,11 @@ unsafe impl Sync for State {}
fn deref(services: &*const Services) -> Option<&Services> {
// SAFETY: We replaced Arc<Services> with *const Services in State. This is
// worth about 10 clones (20 reference count updates) for each request handled.
// Though this is not an incredibly large quantity, it's woefully unnecessary
// given the context as explained below; though it is not currently known to be
// a performance bottleneck, the front-line position justifies preempting it.
// worth about 10 clones (20 reference count updates) for each request
// handled. Though this is not an incredibly large quantity, it's woefully
// unnecessary given the context as explained below; though it is not
// currently known to be a performance bottleneck, the front-line position
// justifies preempting it.
//
// Services is created prior to the axum/tower stack and Router, and prior
// to serving any requests through the handlers in this crate. It is then
+3 -2
View File
@@ -191,8 +191,9 @@ impl Service {
match (&mut list.filters, cached_list.filters.clone()) {
| (Some(filters), Some(cached_filters)) => {
some_or_sticky(&mut filters.is_invite, cached_filters.is_invite);
// TODO (morguldir): Find out how a client can unset this, probably need
// to change into an option inside ruma
// TODO (morguldir): Find out how a client can unset
// this, probably need to change into an option
// inside ruma
list_or_sticky(
&mut filters.not_room_types,
&cached_filters.not_room_types,
Loaded 100 of 113 files, more files were not shown because too many files have changed in this diff. Show more