Compare commits

...

3 commits

Author SHA1 Message Date
nexy7574
ef32fd3482 fix(sync/v2): Room leaves being omitted incorrectly
Partially borrowed from 85a84f93c7
2025-08-30 20:41:40 -07:00
Jacob Taylor
73ca81886a process edus before pdus here, too 2025-08-30 20:41:39 -07:00
Jacob Taylor
9f08594136 pass and use transaction id to collect timing info 2025-08-30 20:41:38 -07:00
2 changed files with 26 additions and 16 deletions

View file

@ -430,7 +430,7 @@ async fn handle_left_room(
.ok(); .ok();
// Left before last sync // Left before last sync
if Some(since) >= left_count { if (Some(since) >= left_count && !include_leave) || Some(next_batch) < left_count {
return Ok(None); return Ok(None);
} }

View file

@ -79,21 +79,14 @@ pub(crate) async fn send_transaction_message_route(
} }
let txn_start_time = Instant::now(); let txn_start_time = Instant::now();
let transaction_id = body.transaction_id.to_string();
trace!( trace!(
pdus = body.pdus.len(), pdus = body.pdus.len(),
edus = body.edus.len(), edus = body.edus.len(),
id = ?body.transaction_id, id = transaction_id,
"Processing transaction", "Processing transaction",
); );
let pdus = body
.pdus
.iter()
.stream()
.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu))
.inspect_err(|e| debug_warn!("Could not parse PDU: {e}"))
.ready_filter_map(Result::ok);
let edus = body let edus = body
.edus .edus
.iter() .iter()
@ -102,21 +95,29 @@ pub(crate) async fn send_transaction_message_route(
.filter_map(Result::ok) .filter_map(Result::ok)
.stream(); .stream();
let pdus = body
.pdus
.iter()
.stream()
.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu))
.inspect_err(|e| debug_warn!("Could not parse PDU: {e}"))
.ready_filter_map(Result::ok);
trace!( trace!(
pdus = body.pdus.len(), pdus = body.pdus.len(),
edus = body.edus.len(), edus = body.edus.len(),
elapsed = ?txn_start_time.elapsed(), elapsed = ?txn_start_time.elapsed(),
id = ?body.transaction_id, id = transaction_id,
"Validated transaction", "Validated transaction",
); );
let results = handle(&services, &client, body.origin(), txn_start_time, pdus, edus).await?; let results = handle(&services, &client, body.origin(), txn_start_time, pdus, edus, &*transaction_id).await?;
info!( info!(
pdus = body.pdus.len(), pdus = body.pdus.len(),
edus = body.edus.len(), edus = body.edus.len(),
elapsed = ?txn_start_time.elapsed(), elapsed = ?txn_start_time.elapsed(),
id = ?body.transaction_id, id = transaction_id,
"Finished txn", "Finished txn",
); );
for (id, result) in &results { for (id, result) in &results {
@ -142,11 +143,13 @@ async fn handle(
started: Instant, started: Instant,
pdus: impl Stream<Item = Pdu> + Send, pdus: impl Stream<Item = Pdu> + Send,
edus: impl Stream<Item = Edu> + Send, edus: impl Stream<Item = Edu> + Send,
transaction_id: &str,
) -> Result<ResolvedMap> { ) -> Result<ResolvedMap> {
let handle_start = Instant::now();
edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu)) edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
.boxed() .boxed()
.await; .await;
let pdu_start = Instant::now();
// group pdus by room // group pdus by room
let pdus = pdus let pdus = pdus
.collect() .collect()
@ -157,7 +160,7 @@ async fn handle(
.collect() .collect()
}) })
.await; .await;
let results_start = Instant::now();
// we can evaluate rooms concurrently // we can evaluate rooms concurrently
let results: ResolvedMap = pdus let results: ResolvedMap = pdus
.into_iter() .into_iter()
@ -172,7 +175,14 @@ async fn handle(
.try_collect() .try_collect()
.boxed() .boxed()
.await?; .await?;
let handle_stop = Instant::now();
info!(
edus = pdu_start.saturating_duration_since(handle_start).as_micros(),
pdus = results_start.saturating_duration_since(pdu_start).as_micros(),
handle_room = handle_stop.saturating_duration_since(results_start).as_micros(),
id = ?transaction_id,
"handled incoming transaction",
);
Ok(results) Ok(results)
} }