Merge branch 'fixallthebugs' into 'next'

fix: all the e2ee problems

See merge request famedly/conduit!393

src/api/appservice_server.rs

@@ -45,11 +45,21 @@ where
     *reqwest_request.timeout_mut() = Some(Duration::from_secs(30));
 
     let url = reqwest_request.url().clone();
-    let mut response = services()
+    let mut response = match services()
         .globals
         .default_client()
         .execute(reqwest_request)
-        .await?;
+        .await
+    {
+        Ok(r) => r,
+        Err(e) => {
+            warn!(
+                "Could not send request to appservice {:?} at {}: {}",
+                registration.get("id"), destination, e
+            );
+            return Err(e.into());
+        }
+    };
 
     // reqwest::Response -> http::Response conversion
     let status = response.status();

src/api/client_server/to_device.rs

@@ -35,6 +35,7 @@ pub async fn send_event_to_device_route(
                 map.insert(target_device_id_maybe.clone(), event.clone());
                 let mut messages = BTreeMap::new();
                 messages.insert(target_user_id.clone(), map);
+                let count = services().globals.next_count()?;
 
                 services().sending.send_reliable_edu(
                     target_user_id.server_name(),
@@ -42,12 +43,12 @@ pub async fn send_event_to_device_route(
                         DirectDeviceContent {
                             sender: sender_user.clone(),
                             ev_type: ToDeviceEventType::from(&*body.event_type),
-                            message_id: body.txn_id.to_owned(),
+                            message_id: count.to_string().into(),
                             messages,
                         },
                     ))
                     .expect("DirectToDevice EDU can be serialized"),
-                    services().globals.next_count()?,
+                    count,
                 )?;
 
                 continue;

src/api/ruma_wrapper/axum.rs

@@ -281,7 +281,7 @@ where
         debug!("{:?}", http_request);
 
         let body = T::try_from_http_request(http_request, &path_params).map_err(|e| {
-            warn!("{:?}", e);
+            warn!("{:?}\n{:?}", e, json_body);
             Error::BadRequest(ErrorKind::BadJson, "Failed to deserialize request.")
         })?;
 

src/api/server_server.rs

@@ -304,7 +304,10 @@ where
                 ))
             }
         }
-        Err(e) => Err(e.into()),
+        Err(e) => {
+            warn!("Could not send request to {} at {}: {}", destination, actual_destination_str, e);
+            Err(e.into())
+        },
     }
 }
 
@@ -831,7 +834,8 @@ pub async fn send_transaction_message_route(
                                     target_user_id,
                                     target_device_id,
                                     &ev_type.to_string(),
-                                    event.deserialize_as().map_err(|_| {
+                                    event.deserialize_as().map_err(|e| {
+                                        warn!("To-Device event is invalid: {event:?} {e}");
                                         Error::BadRequest(
                                             ErrorKind::InvalidParam,
                                             "Event is invalid",

src/database/key_value/sending.rs

@@ -6,7 +6,7 @@ use crate::{
         self,
         sending::{OutgoingKind, SendingEventType},
     },
-    utils, Error, Result,
+    utils, Error, Result, services,
 };
 
 impl service::sending::Data for KeyValueDatabase {
@@ -66,11 +66,11 @@ impl service::sending::Data for KeyValueDatabase {
         let mut keys = Vec::new();
         for (outgoing_kind, event) in requests {
             let mut key = outgoing_kind.get_prefix();
-            key.extend_from_slice(if let SendingEventType::Pdu(value) = &event {
-                value
+            if let SendingEventType::Pdu(value) = &event {
+                key.extend_from_slice(value)
             } else {
-                &[]
-            });
+                key.extend_from_slice(&services().globals.next_count()?.to_be_bytes())
+            }
             let value = if let SendingEventType::Edu(value) = &event {
                 &**value
             } else {

src/service/pusher/mod.rs

@@ -128,7 +128,10 @@ impl Service {
                     Error::BadServerResponse("Push gateway returned bad response.")
                 })
             }
-            Err(e) => Err(e.into()),
+            Err(e) => {
+                warn!("Could not send request to pusher {}: {}", destination, e);
+                Err(e.into())
+            },
         }
     }
 

src/service/sending/mod.rs

@@ -158,8 +158,6 @@ impl Service {
                             // Find events that have been added since starting the last request
                             let new_events = self.db.queued_requests(&outgoing_kind).filter_map(|r| r.ok()).take(30).collect::<Vec<_>>();
 
-                            // TODO: find edus
-
                             if !new_events.is_empty() {
                                 // Insert pdus we found
                                 self.db.mark_as_active(&new_events)?;