diff --git a/src/api/server/send.rs b/src/api/server/send.rs index e7e4e9fd..9c5bfd2b 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -5,7 +5,7 @@ use axum_client_ip::InsecureClientIp; use conduwuit::{ Err, Error, Result, debug, debug::INFO_SPAN_LEVEL, - debug_warn, err, error, info, + debug_warn, err, error, result::LogErr, trace, utils::{ @@ -79,11 +79,13 @@ pub(crate) async fn send_transaction_message_route( } let txn_start_time = Instant::now(); - info!( + trace!( pdus = body.pdus.len(), edus = body.edus.len(), + elapsed = ?txn_start_time.elapsed(), id = ?body.transaction_id, - "Processing transaction", + origin =?body.origin(), + "Starting txn", ); let pdus = body @@ -102,21 +104,14 @@ pub(crate) async fn send_transaction_message_route( .filter_map(Result::ok) .stream(); - info!( - pdus = body.pdus.len(), - edus = body.edus.len(), - elapsed = ?txn_start_time.elapsed(), - id = ?body.transaction_id, - "Validated transaction", - ); - let results = handle(&services, &client, body.origin(), txn_start_time, pdus, edus).await?; - info!( + debug!( pdus = body.pdus.len(), edus = body.edus.len(), elapsed = ?txn_start_time.elapsed(), id = ?body.transaction_id, + origin =?body.origin(), "Finished txn", ); for (id, result) in &results { @@ -143,10 +138,6 @@ async fn handle( pdus: impl Stream + Send, edus: impl Stream + Send, ) -> Result { - edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu)) - .boxed() - .await; - // group pdus by room let pdus = pdus .collect() @@ -163,8 +154,7 @@ async fn handle( .into_iter() .try_stream() .broad_and_then(|(room_id, pdus): (_, Vec<_>)| { - let count = pdus.len(); - handle_room(services, client, origin, started, room_id, pdus.into_iter(), count) + handle_room(services, client, origin, started, room_id, pdus.into_iter()) .map_ok(Vec::into_iter) .map_ok(IterStream::try_stream) }) @@ -173,6 +163,11 @@ async fn handle( .boxed() .await?; + // evaluate edus after pdus, at least for now. + edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu)) + .boxed() + .await; + Ok(results) } @@ -183,7 +178,6 @@ async fn handle_room( txn_start_time: Instant, room_id: OwnedRoomId, pdus: impl Iterator + Send, - count: usize, ) -> Result> { let _room_lock = services .rooms @@ -193,20 +187,10 @@ async fn handle_room( .await; let room_id = &room_id; - let mut n = 0; pdus.try_stream() .and_then(|(_, event_id, value)| async move { services.server.check_running()?; let pdu_start_time = Instant::now(); - info!( - %room_id, - %event_id, - pdu = n + 1, - total = count, - pdu_elapsed = ?pdu_start_time.elapsed(), - txn_elapsed = ?txn_start_time.elapsed(), - "Handling PDU", - ); let result = services .rooms .event_handler @@ -214,16 +198,11 @@ async fn handle_room( .await .map(|_| ()); - info!( - %room_id, - %event_id, - pdu = n + 1, - total = count, + debug!( pdu_elapsed = ?pdu_start_time.elapsed(), txn_elapsed = ?txn_start_time.elapsed(), - "Finished handling PDU {event_id}", + "Finished PDU {event_id}", ); - n += 1; Ok((event_id, result)) })