Compare commits

..

1 Commits

Author SHA1 Message Date
Jade Ellis
5bae8b4f6c feat: Deal with low screen widths in CSS 2025-05-21 02:54:03 +01:00
37 changed files with 663 additions and 852 deletions

View File

@@ -19,20 +19,11 @@ outputs:
rustc_version:
description: The rustc version installed
value: ${{ steps.rustc-version.outputs.version }}
rustup_version:
description: The rustup version installed
value: ${{ steps.rustup-version.outputs.version }}
runs:
using: composite
steps:
- name: Check if rustup is already installed
shell: bash
id: rustup-version
run: |
echo "version=$(rustup --version)" >> $GITHUB_OUTPUT
- name: Cache rustup toolchains
if: steps.rustup-version.outputs.version == ''
uses: actions/cache@v3
with:
path: |
@@ -42,7 +33,6 @@ runs:
# Requires repo to be cloned if toolchain is not specified
key: ${{ runner.os }}-rustup-${{ inputs.toolchain || hashFiles('**/rust-toolchain.toml') }}
- name: Install Rust toolchain
if: steps.rustup-version.outputs.version == ''
shell: bash
run: |
if ! command -v rustup &> /dev/null ; then

View File

@@ -57,6 +57,7 @@ jobs:
build-image:
runs-on: dind
container: ghcr.io/catthehacker/ubuntu:act-latest
needs: define-variables
permissions:
contents: read
@@ -180,14 +181,14 @@ jobs:
file: "docker/Dockerfile"
build-args: |
GIT_COMMIT_HASH=${{ github.sha }})
GIT_COMMIT_HASH_SHORT=${{ env.COMMIT_SHORT_SHA }}
GIT_COMMIT_HASH_SHORT=${{ env.COMMIT_SHORT_SHA }})
GIT_REMOTE_URL=${{github.event.repository.html_url }}
GIT_REMOTE_COMMIT_URL=${{github.event.head_commit.url }}
platforms: ${{ matrix.platform }}
labels: ${{ steps.meta.outputs.labels }}
annotations: ${{ steps.meta.outputs.annotations }}
cache-from: type=gha
# cache-to: type=gha,mode=max
cache-to: type=gha,mode=max
sbom: true
outputs: type=image,"name=${{ needs.define-variables.outputs.images_list }}",push-by-digest=true,name-canonical=true,push=true
env:
@@ -210,6 +211,7 @@ jobs:
merge:
runs-on: dind
container: ghcr.io/catthehacker/ubuntu:act-latest
needs: [define-variables, build-image]
steps:
- name: Download digests

View File

@@ -80,7 +80,6 @@ jobs:
-D warnings
- name: Show sccache stats
if: always()
run: sccache --show-stats
cargo-test:
@@ -138,5 +137,4 @@ jobs:
--no-fail-fast
- name: Show sccache stats
if: always()
run: sccache --show-stats

478
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,63 +0,0 @@
# Security Policy for Continuwuity
This document outlines the security policy for Continuwuity. Our goal is to maintain a secure platform for all users, and we take security matters seriously.
## Supported Versions
We provide security updates for the following versions of Continuwuity:
| Version | Supported |
| -------------- |:----------------:|
| Latest release | ✅ |
| Main branch | ✅ |
| Older releases | ❌ |
We may backport fixes to the previous release at our discretion, but we don't guarantee this.
## Reporting a Vulnerability
### Responsible Disclosure
We appreciate the efforts of security researchers and the community in identifying and reporting vulnerabilities. To ensure that potential vulnerabilities are addressed properly, please follow these guidelines:
1. **Contact members of the team directly** over E2EE private message.
- [@jade:ellis.link](https://matrix.to/#/@jade:ellis.link)
- [@nex:nexy7574.co.uk](https://matrix.to/#/@nex:nexy7574.co.uk) <!-- ? -->
2. **Email the security team** at [security@continuwuity.org](mailto:security@continuwuity.org). This is not E2EE, so don't include sensitive details.
3. **Do not disclose the vulnerability publicly** until it has been addressed
4. **Provide detailed information** about the vulnerability, including:
- A clear description of the issue
- Steps to reproduce
- Potential impact
- Any possible mitigations
- Version(s) affected, including specific commits if possible
If you have any doubts about a potential security vulnerability, contact us via private channels first! We'd prefer that you bother us, instead of having a vulnerability disclosed without a fix.
### What to Expect
When you report a security vulnerability:
1. **Acknowledgment**: We will acknowledge receipt of your report.
2. **Assessment**: We will assess the vulnerability and determine its impact on our users
3. **Updates**: We will provide updates on our progress in addressing the vulnerability, and may request you help test mitigations
4. **Resolution**: Once resolved, we will notify you and discuss coordinated disclosure
5. **Credit**: We will recognize your contribution (unless you prefer to remain anonymous)
## Security Update Process
When security vulnerabilities are identified:
1. We will develop and test fixes in a private fork
2. Security updates will be released as soon as possible
3. Release notes will include information about the vulnerabilities, avoiding details that could facilitate exploitation where possible
4. Critical security updates may be backported to the previous stable release
## Additional Resources
- [Matrix Security Disclosure Policy](https://matrix.org/security-disclosure-policy/)
- [Continuwuity Documentation](https://continuwuity.org/introduction)
---
This security policy was last updated on May 25, 2025.

View File

@@ -1641,29 +1641,19 @@
#
#server =
# URL to a support page for the server, which will be served as part of
# the MSC1929 server support endpoint at /.well-known/matrix/support.
# Will be included alongside any contact information
# This item is undocumented. Please contribute documentation for it.
#
#support_page =
# Role string for server support contacts, to be served as part of the
# MSC1929 server support endpoint at /.well-known/matrix/support.
# This item is undocumented. Please contribute documentation for it.
#
#support_role = "m.role.admin"
#support_role =
# Email address for server support contacts, to be served as part of the
# MSC1929 server support endpoint.
# This will be used along with support_mxid if specified.
# This item is undocumented. Please contribute documentation for it.
#
#support_email =
# Matrix ID for server support contacts, to be served as part of the
# MSC1929 server support endpoint.
# This will be used along with support_email if specified.
#
# If no email or mxid is specified, all of the server's admins will be
# listed.
# This item is undocumented. Please contribute documentation for it.
#
#support_mxid =

View File

@@ -20,4 +20,3 @@ # Summary
- [Testing](development/testing.md)
- [Hot Reloading ("Live" Development)](development/hot_reload.md)
- [Community (and Guidelines)](community.md)
- [Security](security.md)

View File

@@ -1 +0,0 @@
{{#include ../SECURITY.md}}

View File

@@ -125,13 +125,13 @@ pub(super) enum DebugCommand {
reset: bool,
},
/// - Sign JSON blob
/// - Verify json signatures
///
/// This command needs a JSON blob provided in a Markdown code block below
/// the command.
SignJson,
/// - Verify JSON signatures
/// - Verify json signatures
///
/// This command needs a JSON blob provided in a Markdown code block below
/// the command.

View File

@@ -2162,109 +2162,6 @@ 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)
let join_rule = services.rooms.state_accessor.get_join_rules(room_id).await;
if let JoinRule::KnockRestricted(restricted) = &join_rule {
let restriction_rooms: Vec<_> = restricted
.allow
.iter()
.filter_map(|a| match a {
| AllowRule::RoomMembership(r) => Some(&r.room_id),
| _ => None,
})
.collect();
// Check if the user is in any of the allowed rooms
let mut user_meets_restrictions = false;
for restriction_room_id in &restriction_rooms {
if services
.rooms
.state_cache
.is_joined(sender_user, restriction_room_id)
.await
{
user_meets_restrictions = true;
break;
}
}
// If the user meets the restrictions, try joining instead
if user_meets_restrictions {
debug_info!(
"{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
drop(state_lock);
match join_room_by_id_helper(
services,
sender_user,
room_id,
reason.clone(),
servers,
None,
&None,
)
.await
{
| Ok(_) => return Ok(knock_room::v3::Response::new(room_id.to_owned())),
| Err(e) => {
debug_warn!(
"Failed to convert knock to join for {sender_user} in {room_id}: {e:?}"
);
// Get a new state lock for the remaining knock logic
let new_state_lock = services.rooms.state.mutex.lock(room_id).await;
let server_in_room = services
.rooms
.state_cache
.server_in_room(services.globals.server_name(), room_id)
.await;
let local_knock = server_in_room
|| servers.is_empty()
|| (servers.len() == 1 && services.globals.server_is_ours(&servers[0]));
if local_knock {
knock_room_helper_local(
services,
sender_user,
room_id,
reason,
servers,
new_state_lock,
)
.boxed()
.await?;
} else {
knock_room_helper_remote(
services,
sender_user,
room_id,
reason,
servers,
new_state_lock,
)
.boxed()
.await?;
}
return Ok(knock_room::v3::Response::new(room_id.to_owned()));
},
}
}
} else if !matches!(join_rule, JoinRule::Knock | JoinRule::KnockRestricted(_)) {
debug_warn!(
"{sender_user} attempted to knock on room {room_id} but its join rule is \
{join_rule:?}, not knock or knock_restricted"
);
}
let server_in_room = services
.rooms
.state_cache
@@ -2312,12 +2209,6 @@ async fn knock_room_helper_local(
return Err!(Request(Forbidden("This room does not support knocking.")));
}
// Verify that this room has a valid knock or knock_restricted join rule
let join_rule = services.rooms.state_accessor.get_join_rules(room_id).await;
if !matches!(join_rule, JoinRule::Knock | JoinRule::KnockRestricted(_)) {
return Err!(Request(Forbidden("This room's join rule does not allow knocking.")));
}
let content = RoomMemberEventContent {
displayname: services.users.displayname(sender_user).await.ok(),
avatar_url: services.users.avatar_url(sender_user).await.ok(),

View File

@@ -1,7 +1,7 @@
use axum::extract::State;
use conduwuit::{Err, Event, PduEvent, Result, err};
use conduwuit::{Err, Event, Result, err};
use futures::{FutureExt, TryFutureExt, future::try_join};
use ruma::api::client::{error::ErrorKind, room::get_room_event};
use ruma::api::client::room::get_room_event;
use crate::{Ruma, client::is_ignored_pdu};
@@ -14,7 +14,6 @@ pub(crate) async fn get_room_event_route(
) -> Result<get_room_event::v3::Response> {
let event_id = &body.event_id;
let room_id = &body.room_id;
let sender_user = body.sender_user();
let event = services
.rooms
@@ -34,52 +33,6 @@ pub(crate) async fn get_room_event_route(
return Err!(Request(Forbidden("You don't have permission to view this event.")));
}
let include_unredacted_content = body
.include_unredacted_content // User's file has this field name
.unwrap_or(false);
if include_unredacted_content && event.is_redacted() {
let is_server_admin = services
.users
.is_admin(sender_user)
.map(|is_admin| Ok(is_admin));
let can_redact_privilege = services
.rooms
.state_accessor
.user_can_redact(event_id, sender_user, room_id, false) // federation=false for local check
;
let (is_server_admin, can_redact_privilege) =
try_join(is_server_admin, can_redact_privilege).await?;
if !is_server_admin && !can_redact_privilege {
return Err!(Request(Forbidden(
"You don't have permission to view redacted content.",
)));
}
let pdu_id = match services.rooms.timeline.get_pdu_id(event_id).await {
| Ok(id) => id,
| Err(e) => {
return Err(e);
},
};
let original_content = services
.rooms
.timeline
.get_original_pdu_content(&pdu_id)
.await?;
if let Some(original_content) = original_content {
// If the original content is available, we can return it.
// event.content = to_raw_value(&original_content)?;
event = PduEvent::from_id_val(event_id, original_content)?;
} else {
return Err(conduwuit::Error::BadRequest(
ErrorKind::UnredactedContentDeleted { content_keep_ms: None },
"The original unredacted content is not in the database.",
));
}
}
debug_assert!(
event.event_id() == event_id && event.room_id() == room_id,
"Fetched PDU must match requested"

View File

@@ -40,7 +40,6 @@ pub(crate) async fn get_supported_versions_route(
"v1.11".to_owned(),
],
unstable_features: BTreeMap::from_iter([
("fi.mau.msc2815".to_owned(), true),
("org.matrix.e2e_cross_signing".to_owned(), true),
("org.matrix.msc2285.stable".to_owned(), true), /* private read receipts (https://github.com/matrix-org/matrix-spec-proposals/pull/2285) */
("uk.half-shot.msc2666.query_mutual_rooms".to_owned(), true), /* query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666) */

View File

@@ -1,6 +1,5 @@
use axum::{Json, extract::State, response::IntoResponse};
use conduwuit::{Error, Result};
use futures::StreamExt;
use ruma::api::client::{
discovery::{
discover_homeserver::{self, HomeserverInfo, SlidingSyncProxyInfo},
@@ -18,7 +17,7 @@ pub(crate) async fn well_known_client(
State(services): State<crate::State>,
_body: Ruma<discover_homeserver::Request>,
) -> Result<discover_homeserver::Response> {
let client_url = match services.config.well_known.client.as_ref() {
let client_url = match services.server.config.well_known.client.as_ref() {
| Some(url) => url.to_string(),
| None => return Err(Error::BadRequest(ErrorKind::NotFound, "Not found.")),
};
@@ -34,63 +33,44 @@ pub(crate) async fn well_known_client(
/// # `GET /.well-known/matrix/support`
///
/// Server support contact and support page of a homeserver's domain.
/// Implements MSC1929 for server discovery.
/// If no configuration is set, uses admin users as contacts.
pub(crate) async fn well_known_support(
State(services): State<crate::State>,
_body: Ruma<discover_support::Request>,
) -> Result<discover_support::Response> {
let support_page = services
.server
.config
.well_known
.support_page
.as_ref()
.map(ToString::to_string);
let email_address = services.config.well_known.support_email.clone();
let matrix_id = services.config.well_known.support_mxid.clone();
let role = services.server.config.well_known.support_role.clone();
// support page or role must be either defined for this to be valid
if support_page.is_none() && role.is_none() {
return Err(Error::BadRequest(ErrorKind::NotFound, "Not found."));
}
let email_address = services.server.config.well_known.support_email.clone();
let matrix_id = services.server.config.well_known.support_mxid.clone();
// if a role is specified, an email address or matrix id is required
if role.is_some() && (email_address.is_none() && matrix_id.is_none()) {
return Err(Error::BadRequest(ErrorKind::NotFound, "Not found."));
}
// TODO: support defining multiple contacts in the config
let mut contacts: Vec<Contact> = vec![];
let role_value = services
.config
.well_known
.support_role
.clone()
.unwrap_or_else(|| "m.role.admin".to_owned().into());
if let Some(role) = role {
let contact = Contact { role, email_address, matrix_id };
// Add configured contact if at least one contact method is specified
if email_address.is_some() || matrix_id.is_some() {
contacts.push(Contact {
role: role_value.clone(),
email_address: email_address.clone(),
matrix_id: matrix_id.clone(),
});
}
// Try to add admin users as contacts if no contacts are configured
if contacts.is_empty() {
if let Ok(admin_room) = services.admin.get_admin_room().await {
let admin_users = services.rooms.state_cache.room_members(&admin_room);
let mut stream = admin_users;
while let Some(user_id) = stream.next().await {
// Skip server user
if *user_id == services.globals.server_user {
break;
}
contacts.push(Contact {
role: role_value.clone(),
email_address: None,
matrix_id: Some(user_id.to_owned()),
});
}
}
contacts.push(contact);
}
// support page or role+contacts must be either defined for this to be valid
if contacts.is_empty() && support_page.is_none() {
// No admin room, no configured contacts, and no support page
return Err(Error::BadRequest(ErrorKind::NotFound, "Not found."));
}
@@ -104,9 +84,9 @@ pub(crate) async fn well_known_support(
pub(crate) async fn syncv3_client_server_json(
State(services): State<crate::State>,
) -> Result<impl IntoResponse> {
let server_url = match services.config.well_known.client.as_ref() {
let server_url = match services.server.config.well_known.client.as_ref() {
| Some(url) => url.to_string(),
| None => match services.config.well_known.server.as_ref() {
| None => match services.server.config.well_known.server.as_ref() {
| Some(url) => url.to_string(),
| None => return Err(Error::BadRequest(ErrorKind::NotFound, "Not found.")),
},

View File

@@ -13,13 +13,13 @@ version.workspace = true
build = "build.rs"
# [[bin]]
# path = "main.rs"
# name = "conduwuit_build_metadata"
# name = "conduwuit_build_metadata"
[lib]
path = "mod.rs"
crate-type = [
"rlib",
# "dylib",
"rlib",
# "dylib",
]
[features]
@@ -28,7 +28,7 @@ crate-type = [
[dependencies]
[build-dependencies]
built = { version = "0.8", features = [] }
built = {version = "0.7", features = ["cargo-lock", "dependency-tree"]}
[lints]
workspace = true

View File

@@ -78,13 +78,12 @@ fn main() {
}
// --- Rerun Triggers ---
// 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 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}");
}
println!("cargo:rerun-if-env-changed=GIT_COMMIT_HASH");
println!("cargo:rerun-if-env-changed=GIT_COMMIT_HASH_SHORT");

View File

@@ -12,17 +12,11 @@ pub mod built {
v
} else if let v @ Some(_) = option_env!("CONDUWUIT_VERSION_EXTRA") {
v
} else if let v @ Some(_) = option_env!("CONDUIT_VERSION_EXTRA") {
v
} else {
option_env!("CONDUIT_VERSION_EXTRA")
GIT_COMMIT_HASH_SHORT
};
#[must_use]
pub fn version_tag() -> Option<&'static str> {
VERSION_EXTRA
.filter(|s| !s.is_empty())
.or(GIT_COMMIT_HASH_SHORT)
}
pub static GIT_REMOTE_WEB_URL: Option<&str> = option_env!("GIT_REMOTE_WEB_URL");
pub static GIT_REMOTE_COMMIT_URL: Option<&str> = option_env!("GIT_REMOTE_COMMIT_URL");

View File

@@ -274,10 +274,6 @@ pub fn set_dirty_decay<I: Into<Option<usize>>>(arena: I, decay_ms: isize) -> Res
}
}
pub fn background_thread_enable(enable: bool) -> Result<bool> {
set::<u8>(&mallctl!("background_thread"), enable.into()).map(is_nonzero!())
}
#[inline]
#[must_use]
pub fn is_affine_arena() -> bool { is_percpu_arena() || is_phycpu_arena() }

View File

@@ -219,15 +219,6 @@ pub fn check(config: &Config) -> Result {
));
}
// Check if support contact information is configured
if config.well_known.support_email.is_none() && config.well_known.support_mxid.is_none() {
warn!(
"No support contact information (support_email or support_mxid) is configured in \
the well_known section. Users in the admin room will be automatically listed as \
support contacts in the /.well-known/matrix/support endpoint."
);
}
if config
.url_preview_domain_contains_allowlist
.contains(&"*".to_owned())

View File

@@ -1897,28 +1897,12 @@ pub struct WellKnownConfig {
/// example: "matrix.example.com:443"
pub server: Option<OwnedServerName>,
/// URL to a support page for the server, which will be served as part of
/// the MSC1929 server support endpoint at /.well-known/matrix/support.
/// Will be included alongside any contact information
pub support_page: Option<Url>,
/// Role string for server support contacts, to be served as part of the
/// MSC1929 server support endpoint at /.well-known/matrix/support.
///
/// default: "m.role.admin"
pub support_role: Option<ContactRole>,
/// Email address for server support contacts, to be served as part of the
/// MSC1929 server support endpoint.
/// This will be used along with support_mxid if specified.
pub support_email: Option<String>,
/// Matrix ID for server support contacts, to be served as part of the
/// MSC1929 server support endpoint.
/// This will be used along with support_email if specified.
///
/// If no email or mxid is specified, all of the server's admins will be
/// listed.
pub support_mxid: Option<OwnedUserId>,
}

View File

@@ -26,6 +26,6 @@ pub fn user_agent() -> &'static str { USER_AGENT.get_or_init(init_user_agent) }
fn init_user_agent() -> String { format!("{}/{}", name(), version()) }
fn init_version() -> String {
conduwuit_build_metadata::version_tag()
conduwuit_build_metadata::VERSION_EXTRA
.map_or(SEMANTIC.to_owned(), |extra| format!("{SEMANTIC} ({extra})"))
}

View File

@@ -1,10 +1,18 @@
use std::{
borrow::Borrow,
fmt::{Debug, Display},
hash::Hash,
};
use ruma::{EventId, MilliSecondsSinceUnixEpoch, RoomId, UserId, events::TimelineEventType};
use serde_json::value::RawValue as RawJsonValue;
/// Abstraction of a PDU so users can have their own PDU types.
pub trait Event {
type Id: Clone + Debug + Display + Eq + Ord + Hash + Send + Borrow<EventId>;
/// The `EventId` of this event.
fn event_id(&self) -> &EventId;
fn event_id(&self) -> &Self::Id;
/// The `RoomId` of this event.
fn room_id(&self) -> &RoomId;
@@ -26,18 +34,20 @@ pub trait Event {
/// The events before this event.
// Requires GATs to avoid boxing (and TAIT for making it convenient).
fn prev_events(&self) -> impl DoubleEndedIterator<Item = &EventId> + Send + '_;
fn prev_events(&self) -> impl DoubleEndedIterator<Item = &Self::Id> + Send + '_;
/// All the authenticating events for this event.
// Requires GATs to avoid boxing (and TAIT for making it convenient).
fn auth_events(&self) -> impl DoubleEndedIterator<Item = &EventId> + Send + '_;
fn auth_events(&self) -> impl DoubleEndedIterator<Item = &Self::Id> + Send + '_;
/// If this event is a redaction event this is the event it redacts.
fn redacts(&self) -> Option<&EventId>;
fn redacts(&self) -> Option<&Self::Id>;
}
impl<T: Event> Event for &T {
fn event_id(&self) -> &EventId { (*self).event_id() }
type Id = T::Id;
fn event_id(&self) -> &Self::Id { (*self).event_id() }
fn room_id(&self) -> &RoomId { (*self).room_id() }
@@ -51,13 +61,13 @@ fn content(&self) -> &RawJsonValue { (*self).content() }
fn state_key(&self) -> Option<&str> { (*self).state_key() }
fn prev_events(&self) -> impl DoubleEndedIterator<Item = &EventId> + Send + '_ {
fn prev_events(&self) -> impl DoubleEndedIterator<Item = &Self::Id> + Send + '_ {
(*self).prev_events()
}
fn auth_events(&self) -> impl DoubleEndedIterator<Item = &EventId> + Send + '_ {
fn auth_events(&self) -> impl DoubleEndedIterator<Item = &Self::Id> + Send + '_ {
(*self).auth_events()
}
fn redacts(&self) -> Option<&EventId> { (*self).redacts() }
fn redacts(&self) -> Option<&Self::Id> { (*self).redacts() }
}

View File

@@ -79,7 +79,9 @@ pub fn from_id_val(event_id: &EventId, mut json: CanonicalJsonObject) -> Result<
}
impl Event for Pdu {
fn event_id(&self) -> &EventId { &self.event_id }
type Id = OwnedEventId;
fn event_id(&self) -> &Self::Id { &self.event_id }
fn room_id(&self) -> &RoomId { &self.room_id }
@@ -95,15 +97,15 @@ fn origin_server_ts(&self) -> MilliSecondsSinceUnixEpoch {
fn state_key(&self) -> Option<&str> { self.state_key.as_deref() }
fn prev_events(&self) -> impl DoubleEndedIterator<Item = &EventId> + Send + '_ {
self.prev_events.iter().map(AsRef::as_ref)
fn prev_events(&self) -> impl DoubleEndedIterator<Item = &Self::Id> + Send + '_ {
self.prev_events.iter()
}
fn auth_events(&self) -> impl DoubleEndedIterator<Item = &EventId> + Send + '_ {
self.auth_events.iter().map(AsRef::as_ref)
fn auth_events(&self) -> impl DoubleEndedIterator<Item = &Self::Id> + Send + '_ {
self.auth_events.iter()
}
fn redacts(&self) -> Option<&EventId> { self.redacts.as_deref() }
fn redacts(&self) -> Option<&Self::Id> { self.redacts.as_ref() }
}
/// Prevent derived equality which wouldn't limit itself to event_id

View File

@@ -1,8 +1,8 @@
use ruma::{
events::{
AnyMessageLikeEvent, AnyStateEvent, AnyStrippedStateEvent, AnySyncStateEvent,
AnySyncTimelineEvent, AnyTimelineEvent, StateEvent, room::member::RoomMemberEventContent,
space::child::HierarchySpaceChildEvent,
AnyEphemeralRoomEvent, AnyMessageLikeEvent, AnyStateEvent, AnyStrippedStateEvent,
AnySyncStateEvent, AnySyncTimelineEvent, AnyTimelineEvent, StateEvent,
room::member::RoomMemberEventContent, space::child::HierarchySpaceChildEvent,
},
serde::Raw,
};
@@ -10,6 +10,41 @@
use crate::implement;
/// This only works for events that are also AnyRoomEvents.
#[must_use]
#[implement(super::Pdu)]
pub fn into_any_event(self) -> Raw<AnyEphemeralRoomEvent> {
serde_json::from_value(self.into_any_event_value()).expect("Raw::from_value always works")
}
/// This only works for events that are also AnyRoomEvents.
#[implement(super::Pdu)]
#[must_use]
#[inline]
pub fn into_any_event_value(self) -> JsonValue {
let (redacts, content) = self.copy_redacts();
let mut json = json!({
"content": content,
"type": self.kind,
"event_id": self.event_id,
"sender": self.sender,
"origin_server_ts": self.origin_server_ts,
"room_id": self.room_id,
});
if let Some(unsigned) = &self.unsigned {
json["unsigned"] = json!(unsigned);
}
if let Some(state_key) = &self.state_key {
json["state_key"] = json!(state_key);
}
if let Some(redacts) = &redacts {
json["redacts"] = json!(redacts);
}
json
}
#[implement(super::Pdu)]
#[must_use]
#[inline]
@@ -18,8 +53,7 @@ pub fn into_room_event(self) -> Raw<AnyTimelineEvent> { self.to_room_event() }
#[implement(super::Pdu)]
#[must_use]
pub fn to_room_event(&self) -> Raw<AnyTimelineEvent> {
let value = self.to_room_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.to_room_event_value()).expect("Raw::from_value always works")
}
#[implement(super::Pdu)]
@@ -57,8 +91,8 @@ pub fn into_message_like_event(self) -> Raw<AnyMessageLikeEvent> { self.to_messa
#[implement(super::Pdu)]
#[must_use]
pub fn to_message_like_event(&self) -> Raw<AnyMessageLikeEvent> {
let value = self.to_message_like_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.to_message_like_event_value())
.expect("Raw::from_value always works")
}
#[implement(super::Pdu)]
@@ -96,8 +130,7 @@ pub fn into_sync_room_event(self) -> Raw<AnySyncTimelineEvent> { self.to_sync_ro
#[implement(super::Pdu)]
#[must_use]
pub fn to_sync_room_event(&self) -> Raw<AnySyncTimelineEvent> {
let value = self.to_sync_room_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.to_sync_room_event_value()).expect("Raw::from_value always works")
}
#[implement(super::Pdu)]
@@ -129,8 +162,7 @@ pub fn to_sync_room_event_value(&self) -> JsonValue {
#[implement(super::Pdu)]
#[must_use]
pub fn into_state_event(self) -> Raw<AnyStateEvent> {
let value = self.into_state_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.into_state_event_value()).expect("Raw::from_value always works")
}
#[implement(super::Pdu)]
@@ -157,8 +189,8 @@ pub fn into_state_event_value(self) -> JsonValue {
#[implement(super::Pdu)]
#[must_use]
pub fn into_sync_state_event(self) -> Raw<AnySyncStateEvent> {
let value = self.into_sync_state_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.into_sync_state_event_value())
.expect("Raw::from_value always works")
}
#[implement(super::Pdu)]
@@ -191,8 +223,8 @@ pub fn into_stripped_state_event(self) -> Raw<AnyStrippedStateEvent> {
#[implement(super::Pdu)]
#[must_use]
pub fn to_stripped_state_event(&self) -> Raw<AnyStrippedStateEvent> {
let value = self.to_stripped_state_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.to_stripped_state_event_value())
.expect("Raw::from_value always works")
}
#[implement(super::Pdu)]
@@ -210,8 +242,8 @@ pub fn to_stripped_state_event_value(&self) -> JsonValue {
#[implement(super::Pdu)]
#[must_use]
pub fn into_stripped_spacechild_state_event(self) -> Raw<HierarchySpaceChildEvent> {
let value = self.into_stripped_spacechild_state_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.into_stripped_spacechild_state_event_value())
.expect("Raw::from_value always works")
}
#[implement(super::Pdu)]
@@ -230,8 +262,7 @@ pub fn into_stripped_spacechild_state_event_value(self) -> JsonValue {
#[implement(super::Pdu)]
#[must_use]
pub fn into_member_event(self) -> Raw<StateEvent<RoomMemberEventContent>> {
let value = self.into_member_event_value();
serde_json::from_value(value).expect("Failed to serialize Event value")
serde_json::from_value(self.into_member_event_value()).expect("Raw::from_value always works")
}
#[implement(super::Pdu)]

View File

@@ -52,6 +52,7 @@ fn lexico_topo_sort(c: &mut test::Bencher) {
#[cfg(conduwuit_bench)]
#[cfg_attr(conduwuit_bench, bench)]
fn resolution_shallow_auth_chain(c: &mut test::Bencher) {
let parallel_fetches = 32;
let mut store = TestStore(hashmap! {});
// build up the DAG
@@ -77,6 +78,7 @@ fn resolution_shallow_auth_chain(c: &mut test::Bencher) {
&auth_chain_sets,
&fetch,
&exists,
parallel_fetches,
)
.await
{
@@ -89,6 +91,7 @@ fn resolution_shallow_auth_chain(c: &mut test::Bencher) {
#[cfg(conduwuit_bench)]
#[cfg_attr(conduwuit_bench, bench)]
fn resolve_deeper_event_set(c: &mut test::Bencher) {
let parallel_fetches = 32;
let mut inner = INITIAL_EVENTS();
let ban = BAN_STATE_SET();
@@ -150,6 +153,7 @@ fn resolve_deeper_event_set(c: &mut test::Bencher) {
&auth_chain_sets,
&fetch,
&exists,
parallel_fetches,
)
.await
{
@@ -186,11 +190,7 @@ fn get_events(&self, room_id: &RoomId, event_ids: &[OwnedEventId]) -> Result<Vec
}
/// Returns a Vec of the related auth events to the given `event`.
fn auth_event_ids(
&self,
room_id: &RoomId,
event_ids: Vec<OwnedEventId>,
) -> Result<HashSet<OwnedEventId>> {
fn auth_event_ids(&self, room_id: &RoomId, event_ids: Vec<E::Id>) -> Result<HashSet<E::Id>> {
let mut result = HashSet::new();
let mut stack = event_ids;
@@ -216,8 +216,8 @@ fn auth_event_ids(
fn auth_chain_diff(
&self,
room_id: &RoomId,
event_ids: Vec<Vec<OwnedEventId>>,
) -> Result<Vec<OwnedEventId>> {
event_ids: Vec<Vec<E::Id>>,
) -> Result<Vec<E::Id>> {
let mut auth_chain_sets = vec![];
for ids in event_ids {
// TODO state store `auth_event_ids` returns self in the event ids list
@@ -238,7 +238,7 @@ fn auth_chain_diff(
Ok(auth_chain_sets
.into_iter()
.flatten()
.filter(|id| !common.contains(id))
.filter(|id| !common.contains(id.borrow()))
.collect())
} else {
Ok(vec![])
@@ -565,7 +565,7 @@ fn with_state_key(self, state_key: impl Into<String>) -> (StateEventType, String
mod event {
use ruma::{
EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, UserId,
MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, UserId,
events::{TimelineEventType, pdu::Pdu},
};
use serde::{Deserialize, Serialize};
@@ -574,7 +574,9 @@ mod event {
use super::Event;
impl Event for PduEvent {
fn event_id(&self) -> &EventId { &self.event_id }
type Id = OwnedEventId;
fn event_id(&self) -> &Self::Id { &self.event_id }
fn room_id(&self) -> &RoomId {
match &self.rest {
@@ -630,30 +632,28 @@ fn state_key(&self) -> Option<&str> {
}
}
fn prev_events(&self) -> Box<dyn DoubleEndedIterator<Item = &EventId> + Send + '_> {
fn prev_events(&self) -> Box<dyn DoubleEndedIterator<Item = &Self::Id> + Send + '_> {
match &self.rest {
| Pdu::RoomV1Pdu(ev) =>
Box::new(ev.prev_events.iter().map(|(id, _)| id.as_ref())),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.prev_events.iter().map(AsRef::as_ref)),
| Pdu::RoomV1Pdu(ev) => Box::new(ev.prev_events.iter().map(|(id, _)| id)),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.prev_events.iter()),
#[cfg(not(feature = "unstable-exhaustive-types"))]
| _ => unreachable!("new PDU version"),
}
}
fn auth_events(&self) -> Box<dyn DoubleEndedIterator<Item = &EventId> + Send + '_> {
fn auth_events(&self) -> Box<dyn DoubleEndedIterator<Item = &Self::Id> + Send + '_> {
match &self.rest {
| Pdu::RoomV1Pdu(ev) =>
Box::new(ev.auth_events.iter().map(|(id, _)| id.as_ref())),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.auth_events.iter().map(AsRef::as_ref)),
| Pdu::RoomV1Pdu(ev) => Box::new(ev.auth_events.iter().map(|(id, _)| id)),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.auth_events.iter()),
#[cfg(not(feature = "unstable-exhaustive-types"))]
| _ => unreachable!("new PDU version"),
}
}
fn redacts(&self) -> Option<&EventId> {
fn redacts(&self) -> Option<&Self::Id> {
match &self.rest {
| Pdu::RoomV1Pdu(ev) => ev.redacts.as_deref(),
| Pdu::RoomV3Pdu(ev) => ev.redacts.as_deref(),
| Pdu::RoomV1Pdu(ev) => ev.redacts.as_ref(),
| Pdu::RoomV3Pdu(ev) => ev.redacts.as_ref(),
#[cfg(not(feature = "unstable-exhaustive-types"))]
| _ => unreachable!("new PDU version"),
}

View File

@@ -133,7 +133,7 @@ struct RoomMemberContentFields {
level = "debug",
skip_all,
fields(
event_id = incoming_event.event_id().as_str(),
event_id = incoming_event.event_id().borrow().as_str()
)
)]
pub async fn auth_check<F, Fut, Fetched, Incoming>(
@@ -259,7 +259,7 @@ struct RoomCreateContentFields {
// 3. If event does not have m.room.create in auth_events reject
if !incoming_event
.auth_events()
.any(|id| id == room_create_event.event_id())
.any(|id| id.borrow() == room_create_event.event_id().borrow())
{
warn!("no m.room.create event in auth events");
return Ok(false);
@@ -638,7 +638,7 @@ struct GetThirdPartyInvite {
warn!(?target_user_membership_event_id, "Banned user can't join");
false
} else if (join_rules == JoinRule::Invite
|| room_version.allow_knocking && (join_rules == JoinRule::Knock || matches!(join_rules, JoinRule::KnockRestricted(_))))
|| room_version.allow_knocking && join_rules == JoinRule::Knock)
// If the join_rule is invite then allow if membership state is invite or join
&& (target_user_current_membership == MembershipState::Join
|| target_user_current_membership == MembershipState::Invite)
@@ -1021,11 +1021,11 @@ fn check_redaction(
// 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()
if redaction_event.event_id().borrow().server_name()
== redaction_event
.redacts()
.as_ref()
.and_then(|&id| id.server_name())
.and_then(|&id| id.borrow().server_name())
{
debug!("redaction event allowed via room version 1 rules");
return Ok(true);

View File

@@ -20,7 +20,7 @@
use futures::{Future, FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt, future};
use ruma::{
EventId, Int, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomVersionId,
EventId, Int, MilliSecondsSinceUnixEpoch, RoomVersionId,
events::{
StateEventType, TimelineEventType,
room::member::{MembershipState, RoomMemberEventContent},
@@ -39,7 +39,9 @@
debug, debug_error,
matrix::{event::Event, pdu::StateKey},
trace,
utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, WidebandExt},
utils::stream::{
BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryReadyExt, WidebandExt,
},
warn,
};
@@ -67,6 +69,9 @@
/// * `event_fetch` - Any event not found in the `event_map` will defer to this
/// closure to find the event.
///
/// * `parallel_fetches` - The number of asynchronous fetch requests in-flight
/// for any given operation.
///
/// ## Invariants
///
/// The caller of `resolve` must ensure that all the events are from the same
@@ -77,19 +82,21 @@
pub async fn resolve<'a, E, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, ExistsFut>(
room_version: &RoomVersionId,
state_sets: Sets,
auth_chain_sets: &'a [HashSet<OwnedEventId, Hasher>],
auth_chain_sets: &'a [HashSet<E::Id, Hasher>],
event_fetch: &Fetch,
event_exists: &Exists,
) -> Result<StateMap<OwnedEventId>>
parallel_fetches: usize,
) -> Result<StateMap<E::Id>>
where
Fetch: Fn(OwnedEventId) -> FetchFut + Sync,
Fetch: Fn(E::Id) -> FetchFut + Sync,
FetchFut: Future<Output = Option<E>> + Send,
Exists: Fn(OwnedEventId) -> ExistsFut + Sync,
Exists: Fn(E::Id) -> ExistsFut + Sync,
ExistsFut: Future<Output = bool> + Send,
Sets: IntoIterator<IntoIter = SetIter> + Send,
SetIter: Iterator<Item = &'a StateMap<OwnedEventId>> + Clone + Send,
SetIter: Iterator<Item = &'a StateMap<E::Id>> + Clone + Send,
Hasher: BuildHasher + Send + Sync,
E: Event + Clone + Send + Sync,
E::Id: Borrow<EventId> + Send + Sync,
for<'b> &'b E: Send,
{
debug!("State resolution starting");
@@ -140,8 +147,13 @@ pub async fn resolve<'a, E, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Exis
// Sort the control events based on power_level/clock/event_id and
// outgoing/incoming edges
let sorted_control_levels =
reverse_topological_power_sort(control_events, &all_conflicted, &event_fetch).await?;
let sorted_control_levels = reverse_topological_power_sort(
control_events,
&all_conflicted,
&event_fetch,
parallel_fetches,
)
.await?;
debug!(count = sorted_control_levels.len(), "power events");
trace!(list = ?sorted_control_levels, "sorted power events");
@@ -150,7 +162,7 @@ pub async fn resolve<'a, E, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Exis
// Sequentially auth check each control event.
let resolved_control = iterative_auth_check(
&room_version,
sorted_control_levels.iter().stream().map(AsRef::as_ref),
sorted_control_levels.iter().stream(),
clean.clone(),
&event_fetch,
)
@@ -167,7 +179,7 @@ pub async fn resolve<'a, E, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Exis
// that failed auth
let events_to_resolve: Vec<_> = all_conflicted
.iter()
.filter(|&id| !deduped_power_ev.contains(id))
.filter(|&id| !deduped_power_ev.contains(id.borrow()))
.cloned()
.collect();
@@ -187,7 +199,7 @@ pub async fn resolve<'a, E, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Exis
let mut resolved_state = iterative_auth_check(
&room_version,
sorted_left_events.iter().stream().map(AsRef::as_ref),
sorted_left_events.iter().stream(),
resolved_control, // The control events are added to the final resolved state
&event_fetch,
)
@@ -280,14 +292,16 @@ fn get_auth_chain_diff<Id, Hasher>(
/// earlier (further back in time) origin server timestamp.
#[tracing::instrument(level = "debug", skip_all)]
async fn reverse_topological_power_sort<E, F, Fut>(
events_to_sort: Vec<OwnedEventId>,
auth_diff: &HashSet<OwnedEventId>,
events_to_sort: Vec<E::Id>,
auth_diff: &HashSet<E::Id>,
fetch_event: &F,
) -> Result<Vec<OwnedEventId>>
parallel_fetches: usize,
) -> Result<Vec<E::Id>>
where
F: Fn(OwnedEventId) -> Fut + Sync,
F: Fn(E::Id) -> Fut + Sync,
Fut: Future<Output = Option<E>> + Send,
E: Event + Send + Sync,
E::Id: Borrow<EventId> + Send + Sync,
{
debug!("reverse topological sort of power events");
@@ -297,36 +311,35 @@ async fn reverse_topological_power_sort<E, F, Fut>(
}
// This is used in the `key_fn` passed to the lexico_topo_sort fn
let event_to_pl: HashMap<_, _> = graph
let event_to_pl = graph
.keys()
.cloned()
.stream()
.broad_filter_map(async |event_id| {
let pl = get_power_level_for_sender(&event_id, fetch_event)
.await
.ok()?;
Some((event_id, pl))
.map(|event_id| {
get_power_level_for_sender(event_id.clone(), fetch_event)
.map(move |res| res.map(|pl| (event_id, pl)))
})
.inspect(|(event_id, pl)| {
.buffer_unordered(parallel_fetches)
.ready_try_fold(HashMap::new(), |mut event_to_pl, (event_id, pl)| {
debug!(
event_id = event_id.as_str(),
power_level = i64::from(*pl),
event_id = event_id.borrow().as_str(),
power_level = i64::from(pl),
"found the power level of an event's sender",
);
event_to_pl.insert(event_id.clone(), pl);
Ok(event_to_pl)
})
.collect()
.boxed()
.await;
.await?;
let fetcher = async |event_id: OwnedEventId| {
let event_to_pl = &event_to_pl;
let fetcher = |event_id: E::Id| async move {
let pl = *event_to_pl
.get(&event_id)
.get(event_id.borrow())
.ok_or_else(|| Error::NotFound(String::new()))?;
let ev = fetch_event(event_id)
.await
.ok_or_else(|| Error::NotFound(String::new()))?;
Ok((pl, ev.origin_server_ts()))
};
@@ -463,17 +476,18 @@ fn partial_cmp(&self, other: &Self) -> Option<Ordering> { Some(self.cmp(other))
/// the eventId at the eventId's generation (we walk backwards to `EventId`s
/// most recent previous power level event).
async fn get_power_level_for_sender<E, F, Fut>(
event_id: &EventId,
event_id: E::Id,
fetch_event: &F,
) -> serde_json::Result<Int>
where
F: Fn(OwnedEventId) -> Fut + Sync,
F: Fn(E::Id) -> Fut + Sync,
Fut: Future<Output = Option<E>> + Send,
E: Event + Send,
E::Id: Borrow<EventId> + Send,
{
debug!("fetch event ({event_id}) senders power level");
let event = fetch_event(event_id.to_owned()).await;
let event = fetch_event(event_id).await;
let auth_events = event.as_ref().map(Event::auth_events);
@@ -481,7 +495,7 @@ async fn get_power_level_for_sender<E, F, Fut>(
.into_iter()
.flatten()
.stream()
.broadn_filter_map(5, |aid| fetch_event(aid.to_owned()))
.broadn_filter_map(5, |aid| fetch_event(aid.clone()))
.ready_find(|aev| is_type_and_key(aev, &TimelineEventType::RoomPowerLevels, ""))
.await;
@@ -514,13 +528,14 @@ async fn get_power_level_for_sender<E, F, Fut>(
async fn iterative_auth_check<'a, E, F, Fut, S>(
room_version: &RoomVersion,
events_to_check: S,
unconflicted_state: StateMap<OwnedEventId>,
unconflicted_state: StateMap<E::Id>,
fetch_event: &F,
) -> Result<StateMap<OwnedEventId>>
) -> Result<StateMap<E::Id>>
where
F: Fn(OwnedEventId) -> Fut + Sync,
F: Fn(E::Id) -> Fut + Sync,
Fut: Future<Output = Option<E>> + Send,
S: Stream<Item = &'a EventId> + Send + 'a,
E::Id: Borrow<EventId> + Clone + Eq + Ord + Send + Sync + 'a,
S: Stream<Item = &'a E::Id> + Send + 'a,
E: Event + Clone + Send + Sync,
{
debug!("starting iterative auth check");
@@ -528,7 +543,7 @@ async fn iterative_auth_check<'a, E, F, Fut, S>(
let events_to_check: Vec<_> = events_to_check
.map(Result::Ok)
.broad_and_then(async |event_id| {
fetch_event(event_id.to_owned())
fetch_event(event_id.clone())
.await
.ok_or_else(|| Error::NotFound(format!("Failed to find {event_id}")))
})
@@ -536,16 +551,16 @@ async fn iterative_auth_check<'a, E, F, Fut, S>(
.boxed()
.await?;
let auth_event_ids: HashSet<OwnedEventId> = events_to_check
let auth_event_ids: HashSet<E::Id> = events_to_check
.iter()
.flat_map(|event: &E| event.auth_events().map(ToOwned::to_owned))
.flat_map(|event: &E| event.auth_events().map(Clone::clone))
.collect();
let auth_events: HashMap<OwnedEventId, E> = auth_event_ids
let auth_events: HashMap<E::Id, E> = auth_event_ids
.into_iter()
.stream()
.broad_filter_map(fetch_event)
.map(|auth_event| (auth_event.event_id().to_owned(), auth_event))
.map(|auth_event| (auth_event.event_id().clone(), auth_event))
.collect()
.boxed()
.await;
@@ -566,7 +581,7 @@ async fn iterative_auth_check<'a, E, F, Fut, S>(
let mut auth_state = StateMap::new();
for aid in event.auth_events() {
if let Some(ev) = auth_events.get(aid) {
if let Some(ev) = auth_events.get(aid.borrow()) {
//TODO: synapse checks "rejected_reason" which is most likely related to
// soft-failing
auth_state.insert(
@@ -577,7 +592,7 @@ async fn iterative_auth_check<'a, E, F, Fut, S>(
ev.clone(),
);
} else {
warn!(event_id = aid.as_str(), "missing auth event");
warn!(event_id = aid.borrow().as_str(), "missing auth event");
}
}
@@ -586,7 +601,7 @@ async fn iterative_auth_check<'a, E, F, Fut, S>(
.stream()
.ready_filter_map(|key| Some((key, resolved_state.get(key)?)))
.filter_map(|(key, ev_id)| async move {
if let Some(event) = auth_events.get(ev_id) {
if let Some(event) = auth_events.get(ev_id.borrow()) {
Some((key, event.clone()))
} else {
Some((key, fetch_event(ev_id.clone()).await?))
@@ -618,7 +633,7 @@ async fn iterative_auth_check<'a, E, F, Fut, S>(
// add event to resolved state map
resolved_state.insert(
event.event_type().with_state_key(state_key),
event.event_id().to_owned(),
event.event_id().clone(),
);
},
| Ok(false) => {
@@ -645,14 +660,15 @@ async fn iterative_auth_check<'a, E, F, Fut, S>(
/// level as a parent) will be marked as depth 1. depth 1 is "older" than depth
/// 0.
async fn mainline_sort<E, F, Fut>(
to_sort: &[OwnedEventId],
resolved_power_level: Option<OwnedEventId>,
to_sort: &[E::Id],
resolved_power_level: Option<E::Id>,
fetch_event: &F,
) -> Result<Vec<OwnedEventId>>
) -> Result<Vec<E::Id>>
where
F: Fn(OwnedEventId) -> Fut + Sync,
F: Fn(E::Id) -> Fut + Sync,
Fut: Future<Output = Option<E>> + Send,
E: Event + Clone + Send + Sync,
E::Id: Borrow<EventId> + Clone + Send + Sync,
{
debug!("mainline sort of events");
@@ -672,7 +688,7 @@ async fn mainline_sort<E, F, Fut>(
pl = None;
for aid in event.auth_events() {
let ev = fetch_event(aid.to_owned())
let ev = fetch_event(aid.clone())
.await
.ok_or_else(|| Error::NotFound(format!("Failed to find {aid}")))?;
@@ -718,25 +734,26 @@ async fn mainline_sort<E, F, Fut>(
/// that has an associated mainline depth.
async fn get_mainline_depth<E, F, Fut>(
mut event: Option<E>,
mainline_map: &HashMap<OwnedEventId, usize>,
mainline_map: &HashMap<E::Id, usize>,
fetch_event: &F,
) -> Result<usize>
where
F: Fn(OwnedEventId) -> Fut + Sync,
F: Fn(E::Id) -> Fut + Sync,
Fut: Future<Output = Option<E>> + Send,
E: Event + Send + Sync,
E::Id: Borrow<EventId> + Send + Sync,
{
while let Some(sort_ev) = event {
debug!(event_id = sort_ev.event_id().as_str(), "mainline");
debug!(event_id = sort_ev.event_id().borrow().as_str(), "mainline");
let id = sort_ev.event_id();
if let Some(depth) = mainline_map.get(id) {
if let Some(depth) = mainline_map.get(id.borrow()) {
return Ok(*depth);
}
event = None;
for aid in sort_ev.auth_events() {
let aev = fetch_event(aid.to_owned())
let aev = fetch_event(aid.clone())
.await
.ok_or_else(|| Error::NotFound(format!("Failed to find {aid}")))?;
@@ -751,14 +768,15 @@ async fn get_mainline_depth<E, F, Fut>(
}
async fn add_event_and_auth_chain_to_graph<E, F, Fut>(
graph: &mut HashMap<OwnedEventId, HashSet<OwnedEventId>>,
event_id: OwnedEventId,
auth_diff: &HashSet<OwnedEventId>,
graph: &mut HashMap<E::Id, HashSet<E::Id>>,
event_id: E::Id,
auth_diff: &HashSet<E::Id>,
fetch_event: &F,
) where
F: Fn(OwnedEventId) -> Fut + Sync,
F: Fn(E::Id) -> Fut + Sync,
Fut: Future<Output = Option<E>> + Send,
E: Event + Send + Sync,
E::Id: Borrow<EventId> + Clone + Send + Sync,
{
let mut state = vec![event_id];
while let Some(eid) = state.pop() {
@@ -768,27 +786,26 @@ async fn add_event_and_auth_chain_to_graph<E, F, Fut>(
// Prefer the store to event as the store filters dedups the events
for aid in auth_events {
if auth_diff.contains(aid) {
if !graph.contains_key(aid) {
if auth_diff.contains(aid.borrow()) {
if !graph.contains_key(aid.borrow()) {
state.push(aid.to_owned());
}
graph
.get_mut(&eid)
.expect("We just inserted this at the start of the while loop")
.insert(aid.to_owned());
// We just inserted this at the start of the while loop
graph.get_mut(eid.borrow()).unwrap().insert(aid.to_owned());
}
}
}
}
async fn is_power_event_id<E, F, Fut>(event_id: &EventId, fetch: &F) -> bool
async fn is_power_event_id<E, F, Fut>(event_id: &E::Id, fetch: &F) -> bool
where
F: Fn(OwnedEventId) -> Fut + Sync,
F: Fn(E::Id) -> Fut + Sync,
Fut: Future<Output = Option<E>> + Send,
E: Event + Send,
E::Id: Borrow<EventId> + Send + Sync,
{
match fetch(event_id.to_owned()).await.as_ref() {
match fetch(event_id.clone()).await.as_ref() {
| Some(state) => is_power_event(state),
| _ => false,
}
@@ -892,13 +909,13 @@ async fn test_event_sort() {
let fetcher = |id| ready(events.get(&id).cloned());
let sorted_power_events =
super::reverse_topological_power_sort(power_events, &auth_chain, &fetcher)
super::reverse_topological_power_sort(power_events, &auth_chain, &fetcher, 1)
.await
.unwrap();
let resolved_power = super::iterative_auth_check(
&RoomVersion::V6,
sorted_power_events.iter().map(AsRef::as_ref).stream(),
sorted_power_events.iter().stream(),
HashMap::new(), // unconflicted events
&fetcher,
)
@@ -1283,7 +1300,7 @@ async fn test_event_map_none() {
let ev_map = store.0.clone();
let fetcher = |id| ready(ev_map.get(&id).cloned());
let exists = |id: OwnedEventId| ready(ev_map.get(&*id).is_some());
let exists = |id: <PduEvent as Event>::Id| ready(ev_map.get(&*id).is_some());
let state_sets = [state_at_bob, state_at_charlie];
let auth_chain: Vec<_> = state_sets
@@ -1295,13 +1312,19 @@ async fn test_event_map_none() {
})
.collect();
let resolved =
match super::resolve(&RoomVersionId::V2, &state_sets, &auth_chain, &fetcher, &exists)
.await
{
| Ok(state) => state,
| Err(e) => panic!("{e}"),
};
let resolved = match super::resolve(
&RoomVersionId::V2,
&state_sets,
&auth_chain,
&fetcher,
&exists,
1,
)
.await
{
| Ok(state) => state,
| Err(e) => panic!("{e}"),
};
assert_eq!(expected, resolved);
}
@@ -1406,15 +1429,21 @@ async fn ban_with_auth_chains2() {
})
.collect();
let fetcher = |id: OwnedEventId| ready(ev_map.get(&id).cloned());
let exists = |id: OwnedEventId| ready(ev_map.get(&id).is_some());
let resolved =
match super::resolve(&RoomVersionId::V6, &state_sets, &auth_chain, &fetcher, &exists)
.await
{
| Ok(state) => state,
| Err(e) => panic!("{e}"),
};
let fetcher = |id: <PduEvent as Event>::Id| ready(ev_map.get(&id).cloned());
let exists = |id: <PduEvent as Event>::Id| ready(ev_map.get(&id).is_some());
let resolved = match super::resolve(
&RoomVersionId::V6,
&state_sets,
&auth_chain,
&fetcher,
&exists,
1,
)
.await
{
| Ok(state) => state,
| Err(e) => panic!("{e}"),
};
debug!(
resolved = ?resolved

View File

@@ -133,11 +133,17 @@ pub(crate) async fn do_check(
.collect();
let event_map = &event_map;
let fetch = |id: OwnedEventId| ready(event_map.get(&id).cloned());
let exists = |id: OwnedEventId| ready(event_map.get(&id).is_some());
let resolved =
super::resolve(&RoomVersionId::V6, state_sets, &auth_chain_sets, &fetch, &exists)
.await;
let fetch = |id: <PduEvent as Event>::Id| ready(event_map.get(&id).cloned());
let exists = |id: <PduEvent as Event>::Id| ready(event_map.get(&id).is_some());
let resolved = super::resolve(
&RoomVersionId::V6,
state_sets,
&auth_chain_sets,
&fetch,
&exists,
1,
)
.await;
match resolved {
| Ok(state) => state,
@@ -241,8 +247,8 @@ pub(crate) fn get_event(&self, _: &RoomId, event_id: &EventId) -> Result<E> {
pub(crate) fn auth_event_ids(
&self,
room_id: &RoomId,
event_ids: Vec<OwnedEventId>,
) -> Result<HashSet<OwnedEventId>> {
event_ids: Vec<E::Id>,
) -> Result<HashSet<E::Id>> {
let mut result = HashSet::new();
let mut stack = event_ids;
@@ -578,7 +584,7 @@ pub(crate) fn INITIAL_EDGES() -> Vec<OwnedEventId> {
pub(crate) mod event {
use ruma::{
EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, UserId,
MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, UserId,
events::{TimelineEventType, pdu::Pdu},
};
use serde::{Deserialize, Serialize};
@@ -587,7 +593,9 @@ pub(crate) mod event {
use crate::Event;
impl Event for PduEvent {
fn event_id(&self) -> &EventId { &self.event_id }
type Id = OwnedEventId;
fn event_id(&self) -> &Self::Id { &self.event_id }
fn room_id(&self) -> &RoomId {
match &self.rest {
@@ -644,31 +652,29 @@ fn state_key(&self) -> Option<&str> {
}
#[allow(refining_impl_trait)]
fn prev_events(&self) -> Box<dyn DoubleEndedIterator<Item = &EventId> + Send + '_> {
fn prev_events(&self) -> Box<dyn DoubleEndedIterator<Item = &Self::Id> + Send + '_> {
match &self.rest {
| Pdu::RoomV1Pdu(ev) =>
Box::new(ev.prev_events.iter().map(|(id, _)| id.as_ref())),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.prev_events.iter().map(AsRef::as_ref)),
| Pdu::RoomV1Pdu(ev) => Box::new(ev.prev_events.iter().map(|(id, _)| id)),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.prev_events.iter()),
#[allow(unreachable_patterns)]
| _ => unreachable!("new PDU version"),
}
}
#[allow(refining_impl_trait)]
fn auth_events(&self) -> Box<dyn DoubleEndedIterator<Item = &EventId> + Send + '_> {
fn auth_events(&self) -> Box<dyn DoubleEndedIterator<Item = &Self::Id> + Send + '_> {
match &self.rest {
| Pdu::RoomV1Pdu(ev) =>
Box::new(ev.auth_events.iter().map(|(id, _)| id.as_ref())),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.auth_events.iter().map(AsRef::as_ref)),
| Pdu::RoomV1Pdu(ev) => Box::new(ev.auth_events.iter().map(|(id, _)| id)),
| Pdu::RoomV3Pdu(ev) => Box::new(ev.auth_events.iter()),
#[allow(unreachable_patterns)]
| _ => unreachable!("new PDU version"),
}
}
fn redacts(&self) -> Option<&EventId> {
fn redacts(&self) -> Option<&Self::Id> {
match &self.rest {
| Pdu::RoomV1Pdu(ev) => ev.redacts.as_deref(),
| Pdu::RoomV3Pdu(ev) => ev.redacts.as_deref(),
| Pdu::RoomV1Pdu(ev) => ev.redacts.as_ref(),
| Pdu::RoomV3Pdu(ev) => ev.redacts.as_ref(),
#[allow(unreachable_patterns)]
| _ => unreachable!("new PDU version"),
}

View File

@@ -21,10 +21,7 @@
pub use ::tracing;
pub use config::Config;
pub use error::Error;
pub use info::{
rustc_flags_capture, version,
version::{name, version},
};
pub use info::{rustc_flags_capture, version, version::version};
pub use matrix::{Event, EventTypeExt, PduCount, PduEvent, PduId, RoomVersion, pdu, state_res};
pub use server::Server;
pub use utils::{ctor, dtor, implement, result, result::Result};

View File

@@ -121,15 +121,6 @@ pub(super) fn open_list(db: &Arc<Engine>, maps: &[Descriptor]) -> Result<Maps> {
index_size: 512,
..descriptor::SEQUENTIAL
},
Descriptor {
name: "pduid_originalcontent",
cache_disp: CacheDisp::SharedWith("pduid_pdu"),
key_size_hint: Some(16),
val_size_hint: Some(1520),
block_size: 2048,
index_size: 512,
..descriptor::RANDOM
},
Descriptor {
name: "publicroomids",
..descriptor::RANDOM_SMALL

View File

@@ -15,7 +15,7 @@
#[clap(
about,
long_about = None,
name = conduwuit_core::name(),
name = "conduwuit",
version = conduwuit_core::version(),
)]
pub(crate) struct Args {

View File

@@ -98,7 +98,12 @@ pub(super) fn shutdown(server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
Level::INFO
};
wait_shutdown(server, runtime);
debug!(
timeout = ?SHUTDOWN_TIMEOUT,
"Waiting for runtime..."
);
runtime.shutdown_timeout(SHUTDOWN_TIMEOUT);
let runtime_metrics = server.server.metrics.runtime_interval().unwrap_or_default();
event!(LEVEL, ?runtime_metrics, "Final runtime metrics");
@@ -106,23 +111,13 @@ pub(super) fn shutdown(server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
#[cfg(not(tokio_unstable))]
#[tracing::instrument(name = "stop", level = "info", skip_all)]
pub(super) fn shutdown(server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
wait_shutdown(server, runtime);
}
fn wait_shutdown(_server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
pub(super) fn shutdown(_server: &Arc<Server>, runtime: tokio::runtime::Runtime) {
debug!(
timeout = ?SHUTDOWN_TIMEOUT,
"Waiting for runtime..."
);
runtime.shutdown_timeout(SHUTDOWN_TIMEOUT);
// Join any jemalloc threads so they don't appear in use at exit.
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
conduwuit_core::alloc::je::background_thread_enable(false)
.log_debug_err()
.ok();
}
#[tracing::instrument(

View File

@@ -8,7 +8,7 @@
Error, Result, err, implement,
state_res::{self, StateMap},
trace,
utils::stream::{IterStream, ReadyExt, TryWidebandExt, WidebandExt},
utils::stream::{IterStream, ReadyExt, TryWidebandExt, WidebandExt, automatic_width},
};
use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, future::try_join};
use ruma::{OwnedEventId, RoomId, RoomVersionId};
@@ -112,7 +112,14 @@ pub async fn state_resolution<'a, StateSets>(
{
let event_fetch = |event_id| self.event_fetch(event_id);
let event_exists = |event_id| self.event_exists(event_id);
state_res::resolve(room_version, state_sets, auth_chain_sets, &event_fetch, &event_exists)
.map_err(|e| err!(error!("State resolution failed: {e:?}")))
.await
state_res::resolve(
room_version,
state_sets,
auth_chain_sets,
&event_fetch,
&event_exists,
automatic_width(),
)
.map_err(|e| err!(error!("State resolution failed: {e:?}")))
.await
}

View File

@@ -19,8 +19,6 @@ pub(super) struct Data {
pduid_pdu: Arc<Map>,
userroomid_highlightcount: Arc<Map>,
userroomid_notificationcount: Arc<Map>,
/// Stores the original content of redacted PDUs.
pduid_originalcontent: Arc<Map>,
pub(super) db: Arc<Database>,
services: Services,
}
@@ -40,7 +38,6 @@ pub(super) fn new(args: &crate::Args<'_>) -> Self {
pduid_pdu: db["pduid_pdu"].clone(),
userroomid_highlightcount: db["userroomid_highlightcount"].clone(),
userroomid_notificationcount: db["userroomid_notificationcount"].clone(),
pduid_originalcontent: db["pduid_originalcontent"].clone(), // Initialize new table
db: args.db.clone(),
services: Services {
short: args.depend::<rooms::short::Service>("rooms::short"),
@@ -180,24 +177,6 @@ pub(super) async fn get_pdu_json_from_id(
self.pduid_pdu.get(pdu_id).await.deserialized()
}
/// Stores the original content of a PDU that is about to be redacted.
pub(super) async fn store_redacted_pdu_content(
&self,
pdu_id: &RawPduId,
pdu_json: &CanonicalJsonObject,
) -> Result<()> {
self.pduid_originalcontent.raw_put(pdu_id, Json(pdu_json));
Ok(())
}
/// Returns the original content of a redacted PDU.
pub(super) async fn get_original_pdu_content(
&self,
pdu_id: &RawPduId,
) -> Result<Option<CanonicalJsonObject>> {
self.pduid_originalcontent.get(pdu_id).await.deserialized()
}
pub(super) async fn append_pdu(
&self,
pdu_id: &RawPduId,

View File

@@ -260,25 +260,6 @@ pub async fn replace_pdu(
self.db.replace_pdu(pdu_id, pdu_json, pdu).await
}
/// Stores the content of a to-be redacted pdu.
#[tracing::instrument(skip(self), level = "debug")]
pub async fn store_redacted_pdu_content(
&self,
pdu_id: &RawPduId,
pdu_json: &CanonicalJsonObject,
) -> Result<()> {
self.db.store_redacted_pdu_content(pdu_id, pdu_json).await
}
/// Returns the original content of a redacted PDU.
#[tracing::instrument(skip(self), level = "debug")]
pub async fn get_original_pdu_content(
&self,
pdu_id: &RawPduId,
) -> Result<Option<CanonicalJsonObject>> {
self.db.get_original_pdu_content(pdu_id).await
}
/// Creates a new persisted data unit and adds it to a room.
///
/// By this point the incoming event should be fully authenticated, no auth
@@ -491,7 +472,7 @@ pub async fn append_pdu<'a, Leaves>(
.user_can_redact(redact_id, &pdu.sender, &pdu.room_id, false)
.await?
{
self.redact_pdu(redact_id, pdu, shortroomid, true).await?;
self.redact_pdu(redact_id, pdu, shortroomid).await?;
}
}
},
@@ -504,7 +485,7 @@ pub async fn append_pdu<'a, Leaves>(
.user_can_redact(redact_id, &pdu.sender, &pdu.room_id, false)
.await?
{
self.redact_pdu(redact_id, pdu, shortroomid, true).await?;
self.redact_pdu(redact_id, pdu, shortroomid).await?;
}
}
},
@@ -1052,7 +1033,6 @@ pub async fn redact_pdu(
event_id: &EventId,
reason: &PduEvent,
shortroomid: ShortRoomId,
keep_original_content: bool,
) -> Result {
// TODO: Don't reserialize, keep original json
let Ok(pdu_id) = self.get_pdu_id(event_id).await else {
@@ -1074,19 +1054,6 @@ pub async fn redact_pdu(
let room_version_id = self.services.state.get_room_version(&pdu.room_id).await?;
if keep_original_content && !pdu.is_redacted() {
let original_pdu_json = utils::to_canonical_object(&pdu).map_err(|e| {
err!(Database(error!(
?event_id,
?e,
"Failed to convert PDU to canonical JSON for original content storage"
)))
})?;
self.db
.store_redacted_pdu_content(&pdu_id, &original_pdu_json)
.await?;
}
pdu.redact(&room_version_id, reason)?;
let obj = utils::to_canonical_object(&pdu).map_err(|e| {

View File

@@ -1,68 +1,85 @@
:root {
color-scheme: light;
--font-stack: sans-serif;
color-scheme: light;
--font-stack: sans-serif;
--background-color: #fff;
--text-color: #000;
--background-color: #fff;
--text-color: #000;
--bg: oklch(0.76 0.0854 317.27);
--panel-bg: oklch(0.91 0.042 317.27);
--bg: oklch(0.76 0.0854 317.27);
--panel-bg: oklch(0.91 0.042 317.27);
--name-lightness: 0.45;
--name-lightness: 0.45;
@media (prefers-color-scheme: dark) {
color-scheme: dark;
--text-color: #fff;
--bg: oklch(0.15 0.042 317.27);
--panel-bg: oklch(0.24 0.03 317.27);
@media (prefers-color-scheme: dark) {
color-scheme: dark;
--text-color: #fff;
--bg: oklch(0.15 0.042 317.27);
--panel-bg: oklch(0.24 0.03 317.27);
--name-lightness: 0.8;
}
--name-lightness: 0.8;
}
--c1: oklch(0.44 0.177 353.06);
--c2: oklch(0.59 0.158 150.88);
--c1: oklch(0.44 0.177 353.06);
--c2: oklch(0.59 0.158 150.88);
--normal-font-size: 1rem;
--small-font-size: 0.8rem;
--normal-font-size: 1rem;
--small-font-size: 0.8rem;
}
body {
color: var(--text-color);
font-family: var(--font-stack);
margin: 0;
padding: 0;
display: grid;
place-items: center;
min-height: 100vh;
color: var(--text-color);
font-family: var(--font-stack);
margin: 0;
padding: 0;
display: grid;
place-items: center;
min-height: 100vh;
}
html {
background-color: var(--bg);
background-image: linear-gradient(
70deg,
oklch(from var(--bg) l + 0.2 c h),
oklch(from var(--bg) l - 0.2 c h)
);
font-size: 16px;
background-color: var(--bg);
background-image: linear-gradient(
70deg,
oklch(from var(--bg) l + 0.2 c h),
oklch(from var(--bg) l - 0.2 c h)
);
font-size: 16px;
}
.panel {
width: min(clamp(24rem, 12rem + 40vw, 48rem), 100vw);
border-radius: 15px;
background-color: var(--panel-bg);
padding-inline: 1.5rem;
padding-block: 1rem;
box-shadow: 0 0.25em 0.375em hsla(0, 0%, 0%, 0.1);
width: min(clamp(24rem, 12rem + 40vw, 48rem), calc(100vw - 3rem));
border-radius: 15px;
background-color: var(--panel-bg);
padding-inline: 1.5rem;
padding-block: 1rem;
box-shadow: 0 0.25em 0.375em hsla(0, 0%, 0%, 0.1);
}
@media (max-width: 24rem) {
.panel {
padding-inline: 0.25rem;
width: calc(100vw - 0.5rem);
border-radius: 0;
margin-block-start: 0.2rem;
}
main {
height: 100%;
}
}
footer {
padding-inline: 0.25rem;
height: max(fit-content, 2rem);
}
.project-name {
text-decoration: none;
background: linear-gradient(
130deg,
oklch(from var(--c1) var(--name-lightness) c h),
oklch(from var(--c2) var(--name-lightness) c h)
);
background-clip: text;
color: transparent;
filter: brightness(1.2);
text-decoration: none;
background: linear-gradient(
130deg,
oklch(from var(--c1) var(--name-lightness) c h),
oklch(from var(--c2) var(--name-lightness) c h)
);
background-clip: text;
color: transparent;
filter: brightness(1.2);
}

View File

@@ -6,7 +6,7 @@
response::{Html, IntoResponse, Response},
routing::get,
};
use conduwuit_build_metadata::{GIT_REMOTE_COMMIT_URL, GIT_REMOTE_WEB_URL, version_tag};
use conduwuit_build_metadata::{GIT_REMOTE_COMMIT_URL, GIT_REMOTE_WEB_URL, VERSION_EXTRA};
use conduwuit_service::state;
pub fn build() -> Router<state::State> {

View File

@@ -18,7 +18,7 @@
{%~ block footer ~%}
<footer>
<p>Powered by <a href="https://continuwuity.org">Continuwuity</a>
{%~ if let Some(version_info) = self::version_tag() ~%}
{%~ if let Some(version_info) = VERSION_EXTRA ~%}
{%~ if let Some(url) = GIT_REMOTE_COMMIT_URL.or(GIT_REMOTE_WEB_URL) ~%}
(<a href="{{ url }}">{{ version_info }}</a>)
{%~ else ~%}