Immediately trigger EDU sending after client read receipt

Previously, read receipts would only be forwarded via federation
incidentally when some PDU was later sent to the destination server.
Trigger a send without any event to collect EDUs and get read receipts
out directly.

book/changelog.md

@@ -237,6 +237,8 @@ This will be the first release of Grapevine since it was forked from Conduit
     revalidating every alias. This makes it possible to add/remove aliases when
     some of the existing aliases cannot be validated.
     ([!158](https://gitlab.computer.surgery/matrix/grapevine/-/merge_requests/158))
+28. Fix read receipts not being sent over federation (or only arbitrarily late)
+    ([!162](https://gitlab.computer.surgery/matrix/grapevine/-/merge_requests/162))
 
 ### Added
 

src/api/client_server/read_marker.rs

@@ -95,6 +95,9 @@ pub(crate) async fn set_read_marker_route(
                 room_id: body.room_id.clone(),
             },
         )?;
+        for server in services().rooms.state_cache.room_servers(&body.room_id) {
+            services().sending.trigger_edu_send(&server?)?;
+        }
     }
 
     Ok(Ra(set_read_marker::v3::Response {}))
@@ -159,6 +162,11 @@ pub(crate) async fn create_receipt_route(
                     room_id: body.room_id.clone(),
                 },
             )?;
+            for server in
+                services().rooms.state_cache.room_servers(&body.room_id)
+            {
+                services().sending.trigger_edu_send(&server?)?;
+            }
         }
         create_receipt::v3::ReceiptType::ReadPrivate => {
             let count = services()

src/service/sending.rs

@@ -110,8 +110,10 @@ impl RequestKey {
 
 pub(crate) struct RequestData {
     destination: Destination,
-    event_type: SendingEventType,
-    key: RequestKey,
+    /// The PDU or reliable EDU and its associated key, or `None` if this is
+    /// only a trigger to collect and send EDUs to the destination (e.g.
+    /// read receipts).
+    event: Option<(SendingEventType, RequestKey)>,
     /// Span of the original `send_*()` method call
     requester_span: Span,
 }
@@ -309,7 +311,7 @@ impl Service {
     }
 
     #[tracing::instrument(
-        skip(self, event_type, key, requester_span, current_transaction_status),
+        skip(self, event, requester_span, current_transaction_status),
         fields(
             current_status = ?current_transaction_status.get(&destination),
         ),
@@ -318,8 +320,7 @@ impl Service {
         &self,
         RequestData {
             destination,
-            event_type,
-            key,
+            event,
             requester_span,
         }: RequestData,
         current_transaction_status: &mut TransactionStatusMap,
@@ -329,7 +330,7 @@ impl Service {
 
         match self.select_events(
             &destination,
-            vec![(event_type, key)],
+            event.into_iter().collect(),
             current_transaction_status,
         ) {
             Ok(SelectedEvents::Retries(events)) => {
@@ -585,8 +586,7 @@ impl Service {
         self.sender
             .send(RequestData {
                 destination,
-                event_type,
-                key,
+                event: Some((event_type, key)),
                 requester_span: Span::current(),
             })
             .unwrap();
@@ -616,8 +616,7 @@ impl Service {
             self.sender
                 .send(RequestData {
                     destination: destination.clone(),
-                    event_type,
-                    key,
+                    event: Some((event_type, key)),
                     requester_span: Span::current(),
                 })
                 .unwrap();
@@ -644,8 +643,25 @@ impl Service {
         self.sender
             .send(RequestData {
                 destination,
-                event_type,
-                key,
+                event: Some((event_type, key)),
+                requester_span: Span::current(),
+            })
+            .unwrap();
+
+        Ok(())
+    }
+
+    #[tracing::instrument(skip(self))]
+    pub(crate) fn trigger_edu_send(&self, server: &ServerName) -> Result<()> {
+        if server == services().globals.server_name() {
+            debug!("Ignoring EDU send request to ourselves");
+            return Ok(());
+        }
+        let destination = Destination::Normal(server.to_owned());
+        self.sender
+            .send(RequestData {
+                destination,
+                event: None,
                 requester_span: Span::current(),
             })
             .unwrap();
@@ -670,8 +686,7 @@ impl Service {
         self.sender
             .send(RequestData {
                 destination,
-                event_type,
-                key,
+                event: Some((event_type, key)),
                 requester_span: Span::current(),
             })
             .unwrap();