Compare commits

..

8 commits

Author SHA1 Message Date
Jacob Taylor
83a97e63d2 more funny settings (part 3 of 12)
Some checks failed
Checks / Prefligit / prefligit (push) Failing after 3s
Release Docker Image / define-variables (push) Failing after 2s
Release Docker Image / build-image (linux/amd64, release, linux-amd64, base) (push) Has been skipped
Release Docker Image / build-image (linux/arm64, release, linux-arm64, base) (push) Has been skipped
Release Docker Image / merge (push) Has been skipped
Checks / Rust / Format (push) Failing after 5s
Checks / Rust / Clippy (push) Failing after 41s
Checks / Rust / Cargo Test (push) Failing after 39s
2025-07-03 14:32:05 -07:00
Jacob Taylor
70e15c4e1a sender_workers scaling. this time, with feeling! 2025-07-03 14:32:05 -07:00
Jacob Taylor
3230e1acae vehicle loan documentation now available at window 7
also print event id
2025-07-03 14:32:05 -07:00
Jacob Taylor
d0aa9eade4 lock the getter instead ??? c/o M 2025-07-03 14:32:05 -07:00
Jacob Taylor
2b61f91f1f make fetching key room events less smart 2025-07-03 14:31:50 -07:00
Jacob Taylor
59cd268a3f change rocksdb stats level to 3
scale rocksdb background jobs and subcompactions

change rocksdb default error level to info from error

delete unused num_threads function

fix warns from cargo
2025-07-03 14:31:50 -07:00
nexy7574
95a92e5f59 log which room struggled to get mainline depth 2025-07-03 14:30:26 -07:00
nexy7574
2f643b8fbf more logs 2025-07-03 14:30:26 -07:00
6 changed files with 75 additions and 3902 deletions

File diff suppressed because it is too large Load diff

View file

@ -275,11 +275,9 @@ where
if room_id_server_name != room_create_event.sender().server_name() { if room_id_server_name != room_create_event.sender().server_name() {
warn!( warn!(
"servername of room ID origin ({}) does not match servername of m.room.create \ "servername of room ID origin ({}) does not match servername of m.room.create sender ({})",
sender ({})",
room_id_server_name, room_id_server_name,
room_create_event.sender().server_name() room_create_event.sender().server_name());
);
return Ok(false); return Ok(false);
} }

View file

@ -616,7 +616,7 @@ where
.map(ToOwned::to_owned), .map(ToOwned::to_owned),
) )
}; };
debug!("running auth check on {:?}", event.event_id());
let auth_result = let auth_result =
auth_check(room_version, &event, current_third_party, fetch_state).await; auth_check(room_version, &event, current_third_party, fetch_state).await;
@ -735,7 +735,7 @@ where
{ {
let mut room_id = None; let mut room_id = None;
while let Some(sort_ev) = event { while let Some(sort_ev) = event {
trace!(event_id = sort_ev.event_id().as_str(), "mainline"); debug!(event_id = sort_ev.event_id().as_str(), "mainline");
if room_id.is_none() { if room_id.is_none() {
room_id = Some(sort_ev.room_id().to_owned()); room_id = Some(sort_ev.room_id().to_owned());
} }

View file

@ -80,7 +80,7 @@ where
// 5. Reject "due to auth events" if can't get all the auth events or some of // 5. Reject "due to auth events" if can't get all the auth events or some of
// the auth events are also rejected "due to auth events" // the auth events are also rejected "due to auth events"
// NOTE: Step 5 is not applied anymore because it failed too often // NOTE: Step 5 is not applied anymore because it failed too often
debug!("Fetching auth events for {}", incoming_pdu.event_id); debug!("Fetching auth events");
Box::pin(self.fetch_and_handle_outliers( Box::pin(self.fetch_and_handle_outliers(
origin, origin,
pdu_event.auth_events(), pdu_event.auth_events(),
@ -92,12 +92,12 @@ where
// 6. Reject "due to auth events" if the event doesn't pass auth based on the // 6. Reject "due to auth events" if the event doesn't pass auth based on the
// auth events // auth events
debug!("Checking {} based on auth events", incoming_pdu.event_id); debug!("Checking based on auth events");
// Build map of auth events // Build map of auth events
let mut auth_events = HashMap::with_capacity(pdu_event.auth_events().count()); let mut auth_events = HashMap::with_capacity(pdu_event.auth_events().count());
for id in pdu_event.auth_events() { for id in pdu_event.auth_events() {
let Ok(auth_event) = self.services.timeline.get_pdu(id).await else { let Ok(auth_event) = self.services.timeline.get_pdu(id).await else {
warn!("Could not find auth event {id} for {}", incoming_pdu.event_id); warn!("Could not find auth event {id}");
continue; continue;
}; };
@ -131,7 +131,6 @@ where
ready(auth_events.get(&key).map(ToOwned::to_owned)) ready(auth_events.get(&key).map(ToOwned::to_owned))
}; };
debug!("running auth check to handle outlier pdu {:?}", incoming_pdu.event_id);
let auth_check = state_res::event_auth::auth_check( let auth_check = state_res::event_auth::auth_check(
&to_room_version(&room_version_id), &to_room_version(&room_version_id),
&pdu_event, &pdu_event,

View file

@ -1,6 +1,12 @@
use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::Instant}; use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::Instant};
use conduwuit::{Err, Result, debug, debug_info, err, implement, matrix::{EventTypeExt, PduEvent, StateKey, state_res}, trace, utils::stream::{BroadbandExt, ReadyExt}, warn, info}; use conduwuit::{
Err, Result, debug, debug_info, err, implement, is_equal_to,
matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
trace,
utils::stream::{BroadbandExt, ReadyExt},
warn,
};
use futures::{FutureExt, StreamExt, future::ready}; use futures::{FutureExt, StreamExt, future::ready};
use ruma::{CanonicalJsonValue, RoomId, ServerName, events::StateEventType}; use ruma::{CanonicalJsonValue, RoomId, ServerName, events::StateEventType};
@ -11,19 +17,22 @@ use crate::rooms::{
}; };
#[implement(super::Service)] #[implement(super::Service)]
pub(super) async fn upgrade_outlier_to_timeline_pdu( pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
&self, &self,
incoming_pdu: PduEvent, incoming_pdu: PduEvent,
val: BTreeMap<String, CanonicalJsonValue>, val: BTreeMap<String, CanonicalJsonValue>,
create_event: &PduEvent, create_event: &Pdu,
origin: &ServerName, origin: &ServerName,
room_id: &RoomId, room_id: &RoomId,
) -> Result<Option<RawPduId>> { ) -> Result<Option<RawPduId>>
where
Pdu: Event + Send + Sync,
{
// Skip the PDU if we already have it as a timeline event // Skip the PDU if we already have it as a timeline event
if let Ok(pduid) = self if let Ok(pduid) = self
.services .services
.timeline .timeline
.get_pdu_id(&incoming_pdu.event_id) .get_pdu_id(incoming_pdu.event_id())
.await .await
{ {
return Ok(Some(pduid)); return Ok(Some(pduid));
@ -32,13 +41,13 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
if self if self
.services .services
.pdu_metadata .pdu_metadata
.is_event_soft_failed(&incoming_pdu.event_id) .is_event_soft_failed(incoming_pdu.event_id())
.await .await
{ {
return Err!(Request(InvalidParam("Event has been soft failed"))); return Err!(Request(InvalidParam("Event has been soft failed")));
} }
debug!("Upgrading pdu {} from outlier to timeline pdu", incoming_pdu.event_id); debug!("Upgrading to timeline pdu");
let timer = Instant::now(); let timer = Instant::now();
let room_version_id = get_room_version_id(create_event)?; let room_version_id = get_room_version_id(create_event)?;
@ -46,8 +55,8 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
// backwards extremities doing all the checks in this list starting at 1. // backwards extremities doing all the checks in this list starting at 1.
// These are not timeline events. // These are not timeline events.
debug!("Resolving state at event {}", incoming_pdu.event_id); debug!("Resolving state at event");
let mut state_at_incoming_event = if incoming_pdu.prev_events.len() == 1 { let mut state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
self.state_at_incoming_degree_one(&incoming_pdu).await? self.state_at_incoming_degree_one(&incoming_pdu).await?
} else { } else {
self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_id) self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_id)
@ -56,15 +65,16 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
if state_at_incoming_event.is_none() { if state_at_incoming_event.is_none() {
state_at_incoming_event = self state_at_incoming_event = self
.fetch_state(origin, create_event, room_id, &incoming_pdu.event_id) .fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
.await?; .await?;
} }
let state_at_incoming_event = let state_at_incoming_event =
state_at_incoming_event.expect("we always set this to some above"); state_at_incoming_event.expect("we always set this to some above");
let room_version = to_room_version(&room_version_id); let room_version = to_room_version(&room_version_id);
debug!("Performing auth check to upgrade {}", incoming_pdu.event_id); debug!("Performing auth check");
// 11. Check the auth of the event passes based on the state of the event // 11. Check the auth of the event passes based on the state of the event
let state_fetch_state = &state_at_incoming_event; let state_fetch_state = &state_at_incoming_event;
let state_fetch = |k: StateEventType, s: StateKey| async move { let state_fetch = |k: StateEventType, s: StateKey| async move {
@ -74,7 +84,6 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
self.services.timeline.get_pdu(event_id).await.ok() self.services.timeline.get_pdu(event_id).await.ok()
}; };
debug!("running auth check on {}", incoming_pdu.event_id);
let auth_check = state_res::event_auth::auth_check( let auth_check = state_res::event_auth::auth_check(
&room_version, &room_version,
&incoming_pdu, &incoming_pdu,
@ -88,25 +97,24 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
return Err!(Request(Forbidden("Event has failed auth check with state at the event."))); return Err!(Request(Forbidden("Event has failed auth check with state at the event.")));
} }
debug!("Gathering auth events for {}", incoming_pdu.event_id); debug!("Gathering auth events");
let auth_events = self let auth_events = self
.services .services
.state .state
.get_auth_events( .get_auth_events(
room_id, room_id,
&incoming_pdu.kind, incoming_pdu.kind(),
&incoming_pdu.sender, incoming_pdu.sender(),
incoming_pdu.state_key.as_deref(), incoming_pdu.state_key(),
&incoming_pdu.content, incoming_pdu.content(),
) )
.await?; .await?;
let state_fetch = |k: &StateEventType, s: &str| { let state_fetch = |k: &StateEventType, s: &str| {
let key = k.with_state_key(s); let key = k.with_state_key(s);
ready(auth_events.get(&key).cloned()) ready(auth_events.get(&key).map(ToOwned::to_owned))
}; };
debug!("running auth check on {} with claimed state auth", incoming_pdu.event_id);
let auth_check = state_res::event_auth::auth_check( let auth_check = state_res::event_auth::auth_check(
&room_version, &room_version,
&incoming_pdu, &incoming_pdu,
@ -117,7 +125,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?; .map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
// Soft fail check before doing state res // Soft fail check before doing state res
debug!("Performing soft-fail check on {}", incoming_pdu.event_id); debug!("Performing soft-fail check");
let soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_id)) { let soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_id)) {
| (false, _) => true, | (false, _) => true,
| (true, None) => false, | (true, None) => false,
@ -125,7 +133,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
!self !self
.services .services
.state_accessor .state_accessor
.user_can_redact(&redact_id, &incoming_pdu.sender, &incoming_pdu.room_id, true) .user_can_redact(&redact_id, incoming_pdu.sender(), incoming_pdu.room_id(), true)
.await?, .await?,
}; };
@ -145,7 +153,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
.map(ToOwned::to_owned) .map(ToOwned::to_owned)
.ready_filter(|event_id| { .ready_filter(|event_id| {
// Remove any that are referenced by this incoming event's prev_events // Remove any that are referenced by this incoming event's prev_events
!incoming_pdu.prev_events.contains(event_id) !incoming_pdu.prev_events().any(is_equal_to!(event_id))
}) })
.broad_filter_map(|event_id| async move { .broad_filter_map(|event_id| async move {
// Only keep those extremities were not referenced yet // Only keep those extremities were not referenced yet
@ -164,7 +172,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
debug!( debug!(
"Retained {} extremities checked against {} prev_events", "Retained {} extremities checked against {} prev_events",
extremities.len(), extremities.len(),
incoming_pdu.prev_events.len() incoming_pdu.prev_events().count()
); );
let state_ids_compressed: Arc<CompressedState> = self let state_ids_compressed: Arc<CompressedState> = self
@ -179,20 +187,20 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
.map(Arc::new) .map(Arc::new)
.await; .await;
if incoming_pdu.state_key.is_some() { if incoming_pdu.state_key().is_some() {
debug!("Event is a state-event. Deriving new room state"); debug!("Event is a state-event. Deriving new room state");
// We also add state after incoming event to the fork states // We also add state after incoming event to the fork states
let mut state_after = state_at_incoming_event.clone(); let mut state_after = state_at_incoming_event.clone();
if let Some(state_key) = &incoming_pdu.state_key { if let Some(state_key) = incoming_pdu.state_key() {
let shortstatekey = self let shortstatekey = self
.services .services
.short .short
.get_or_create_shortstatekey(&incoming_pdu.kind.to_string().into(), state_key) .get_or_create_shortstatekey(&incoming_pdu.kind().to_string().into(), state_key)
.await; .await;
let event_id = &incoming_pdu.event_id; let event_id = incoming_pdu.event_id();
state_after.insert(shortstatekey, event_id.clone()); state_after.insert(shortstatekey, event_id.to_owned());
} }
let new_room_state = self let new_room_state = self
@ -216,7 +224,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
// 14. Check if the event passes auth based on the "current state" of the room, // 14. Check if the event passes auth based on the "current state" of the room,
// if not soft fail it // if not soft fail it
if soft_fail { if soft_fail {
info!("Soft failing event {}", incoming_pdu.event_id); debug!("Soft failing event");
let extremities = extremities.iter().map(Borrow::borrow); let extremities = extremities.iter().map(Borrow::borrow);
self.services self.services
@ -234,9 +242,9 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
// Soft fail, we keep the event as an outlier but don't add it to the timeline // Soft fail, we keep the event as an outlier but don't add it to the timeline
self.services self.services
.pdu_metadata .pdu_metadata
.mark_event_soft_failed(&incoming_pdu.event_id); .mark_event_soft_failed(incoming_pdu.event_id());
warn!("Event was soft failed: {incoming_pdu:?}"); warn!("Event was soft failed: {:?}", incoming_pdu.event_id());
return Err!(Request(InvalidParam("Event has been soft failed"))); return Err!(Request(InvalidParam("Event has been soft failed")));
} }
@ -247,7 +255,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
let extremities = extremities let extremities = extremities
.iter() .iter()
.map(Borrow::borrow) .map(Borrow::borrow)
.chain(once(incoming_pdu.event_id.borrow())); .chain(once(incoming_pdu.event_id()));
let pdu_id = self let pdu_id = self
.services .services

File diff suppressed because it is too large Load diff