diff --git a/src/api/client/sync/v3.rs b/src/api/client/sync/v3.rs index 01428c08..298a6e4b 100644 --- a/src/api/client/sync/v3.rs +++ b/src/api/client/sync/v3.rs @@ -430,7 +430,7 @@ async fn handle_left_room( .ok(); // Left before last sync - if Some(since) >= left_count { + if (Some(since) >= left_count && !include_leave) || Some(next_batch) < left_count { return Ok(None); } diff --git a/src/api/server/send.rs b/src/api/server/send.rs index 601da06c..cd563b18 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -79,21 +79,14 @@ pub(crate) async fn send_transaction_message_route( } let txn_start_time = Instant::now(); + let transaction_id = body.transaction_id.to_string(); trace!( pdus = body.pdus.len(), edus = body.edus.len(), - id = ?body.transaction_id, + id = transaction_id, "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 .edus .iter() @@ -102,21 +95,29 @@ pub(crate) async fn send_transaction_message_route( .filter_map(Result::ok) .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!( pdus = body.pdus.len(), edus = body.edus.len(), elapsed = ?txn_start_time.elapsed(), - id = ?body.transaction_id, + id = transaction_id, "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!( pdus = body.pdus.len(), edus = body.edus.len(), elapsed = ?txn_start_time.elapsed(), - id = ?body.transaction_id, + id = transaction_id, "Finished txn", ); for (id, result) in &results { @@ -142,11 +143,13 @@ async fn handle( started: Instant, pdus: impl Stream + Send, edus: impl Stream + Send, + transaction_id: &str, ) -> Result { + let handle_start = Instant::now(); edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu)) .boxed() .await; - + let pdu_start = Instant::now(); // group pdus by room let pdus = pdus .collect() @@ -157,7 +160,7 @@ async fn handle( .collect() }) .await; - + let results_start = Instant::now(); // we can evaluate rooms concurrently let results: ResolvedMap = pdus .into_iter() @@ -172,7 +175,14 @@ async fn handle( .try_collect() .boxed() .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) }