macltok

public

Star0

Fork 0

Watch0

24 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `38c4caecdb3d`

tarball

atagen <boss@atagen.co>2026-09-27 01:42 UTC

7 files changed+198-1

engine: warm a room up for sending

The sdk does its member sync, key query, key claim and room key share inline on the send path, so under federation lag a send into an encrypted room can stall for a minute. prepare_to_send syncs an encrypted room's members ahead of time, which has the sdk mark them all tracked and run the /keys/query in its own background requests, deduplicated per room and run by itself on open and on typing unless EngineConfig::prepare_to_send is off. The claim and share stay on the send path until matrix-sdk exposes preshare_room_key.

Modified file: README.md+6 additions@@ -36,6 +36,12 @@ and awaits one of those answers deadlocks once it fills. from the draining
 task, `try_send` the command and drop the `Pending`, the events tell you how it
 went
 
+`prepare_to_send` does the key work a send into an encrypted room would start
+with, ahead of time, so the send itself is quicker. the engine runs it itself
+when a room opens or `set_typing(room, true)` comes in, unless
+`EngineConfig::with_prepare_to_send(false)`. only part of that work is reachable
+without a matrix-sdk fork, `docs/design-decisions.md` says which
+
 start a timeline from `TimelineReset`, then apply each `TimelineChanged` whole
 with `TimelineDiff::try_apply_all`. snapshots are `imbl::Vector`s, re-exported
 as `macltok::imbl`, so a copy costs nothing. when a batch doesn't fit none of it

Modified file: docs/design-decisions.md+30 additions@@ -113,6 +113,36 @@ edit, redact and react take an `EventRef`, so your own message can be edited or
 reacted to before the server has echoed it back. redacting an unsent one just
 cancels it
 
+## warming up to send
+
+the sdk does all its key work on the send path, in the room's send queue before
+the `/send` itself. members synced, a `/keys/query` for anyone untracked or
+dirty, then a `/keys/claim` and the room key sent to every device. over a slow
+federation that's a minute of nothing. `prepare_to_send` does what the public
+api lets us of that, ahead of time, which comes down to syncing the members if
+they aren't. the sdk dedups that with its own sync, and it marks every member
+tracked and dirty, so the sdk's background outgoing requests run the
+`/keys/query` long before the send would have. that's what moves the query off
+the send path. we don't ask `request_user_identity` for anyone, since that's
+an out of band query that never marks a user tracked or up to date
+(matrix-sdk-crypto `identities/manager.rs` ~804-817), federated traffic the
+send path would redo anyway. a synced room costs nothing, so there's no
+cooldown. one warm-up per room at a time, a second ask joins it. opening a room
+or starting to type runs one too, unless `EngineConfig::prepare_to_send` is
+off. it's on because malice asked, not because anyone decided
+
+that's the leftovers of turning down a matrix-sdk fork. the `/keys/claim` and
+the room key share, the slow federated half, stay on the send path. the
+background `/keys/query` runs when the sdk next sends its outgoing requests,
+after a sync response, and `send_outgoing_requests` is crate private, so a send
+straight after a warm-up can still find the members dirty and query them itself.
+and we can't join the dedup the sdk wraps its own prep in
+
+the hook upstream is the todo above `preshare_room_key` at matrix-sdk
+`room/mod.rs` ~2392, asking for exactly this. it's still private on main and in
+0.19.1 as of 2026-09-26. once upstream makes it public, or a fork becomes
+acceptable, `prepare_to_send` should just call it and most of this goes
+
 ## mutations
 
 membership, moderation, power levels, room creation and account data writes go

Modified file: src/engine/mod.rs+33 additions−1 deletions@@ -2,6 +2,7 @@ mod commands;
 pub(crate) mod executor;
 mod media;
 mod mutations;
+mod prepare;
 mod room;
 mod session;
 mod timeline;
@@ -46,6 +47,8 @@ pub struct EngineConfig {
     pub initial_history_limit: u16,
     pub max_media_bytes: usize,
     pub max_media_sources: usize,
+    /// warm a room up for sending when it's opened or someone starts typing in it
+    pub prepare_to_send: bool,
 }
 
 impl EngineConfig {
@@ -78,6 +81,12 @@ impl EngineConfig {
         self.max_media_sources = sources;
         self
     }
+
+    #[must_use]
+    pub const fn with_prepare_to_send(mut self, prepare: bool) -> Self {
+        self.prepare_to_send = prepare;
+        self
+    }
 }
 
 impl Default for EngineConfig {
@@ -88,6 +97,7 @@ impl Default for EngineConfig {
             initial_history_limit: 30,
             max_media_bytes: 20 * 1024 * 1024,
             max_media_sources: 2_048,
+            prepare_to_send: true,
         }
     }
 }
@@ -262,6 +272,7 @@ pub struct Engine {
     stop: mpsc::UnboundedSender<Termination>,
     stops: mpsc::UnboundedReceiver<Termination>,
     logouts: Arc<watch::Sender<Logouts>>,
+    preparations: prepare::Preparations,
 }
 
 impl Engine {
@@ -345,6 +356,7 @@ impl Engine {
             upload_reports,
             uploads: HashMap::new(),
             verification_handler: None,
+            preparations: prepare::Preparations::default(),
             stop,
             stops,
             logouts: Arc::new(watch::Sender::new(Logouts::default())),
@@ -640,9 +652,21 @@ impl Engine {
                     commands::delete_devices(&client, &device_ids, auth).await
                 });
             }
-            Command::Room(RoomCommand::Open(room_id)) => self.open_room(room_id, reply).await,
+            Command::Room(RoomCommand::Open(room_id)) => {
+                if self.config.prepare_to_send {
+                    self.prepare_quietly(room_id.clone());
+                }
+                self.open_room(room_id, reply).await;
+            }
             Command::Room(RoomCommand::Close(room_id)) => self.close_room(room_id, reply).await,
+            Command::Room(RoomCommand::PrepareToSend(room_id)) => {
+                let flight = self.preparations.join(&client, room_id);
+                self.spawn_command("prepare to send", reply, flight);
+            }
             Command::Room(RoomCommand::SetTyping { room_id, typing }) => {
+                if typing && self.config.prepare_to_send {
+                    self.prepare_quietly(room_id.clone());
+                }
                 self.spawn_command("typing", reply, async move {
                     commands::set_typing(&client, &room_id, typing).await
                 });
@@ -687,6 +711,14 @@ impl Engine {
         });
     }
 
+    fn prepare_quietly(&mut self, room_id: OwnedRoomId) {
+        let flight = self.preparations.join(&self.client, room_id);
+        self.tasks
+            .spawn(TaskContext::Command("prepare to send"), async move {
+                let _ = flight.await;
+            });
+    }
+
     fn upload(&mut self, upload: crate::AttachmentUpload, reply: Responder) {
         let Some(room) = self.client.get_room(upload.timeline_id.room_id()) else {
             return reply.done(Err(Error::room_unknown()));

Added file: src/engine/prepare.rs+47 additions@@ -0,0 +1,47 @@
+use std::collections::HashMap;
+
+use futures_util::{
+    FutureExt,
+    future::{Shared, WeakShared},
+};
+use matrix_sdk::Client;
+use ruma::OwnedRoomId;
+
+use super::executor::EngineBoxFuture;
+use crate::Error;
+
+type Flight = Shared<EngineBoxFuture<'static, Result<(), Error>>>;
+
+/// one warm-up per room at a time, later asks join the one in flight
+#[derive(Default)]
+pub(super) struct Preparations(
+    HashMap<OwnedRoomId, WeakShared<EngineBoxFuture<'static, Result<(), Error>>>>,
+);
+
+impl Preparations {
+    pub(super) fn join(&mut self, client: &Client, room_id: OwnedRoomId) -> Flight {
+        self.0.retain(|_, flight| flight.upgrade().is_some());
+        if let Some(flight) = self.0.get(&room_id).and_then(WeakShared::upgrade) {
+            return flight;
+        }
+        let work: EngineBoxFuture<'static, _> = Box::pin(prepare(client.clone(), room_id.clone()));
+        let flight = work.shared();
+        if let Some(weak) = flight.downgrade() {
+            self.0.insert(room_id, weak);
+        }
+        flight
+    }
+}
+
+/// the part of the sdk's pre-send encryption work it lets us reach. syncing the members
+/// has the sdk track them all, and its own background `/keys/query` then catches up
+/// their devices. the one time key claim and the room key share stay on the send path
+async fn prepare(client: Client, room_id: OwnedRoomId) -> Result<(), Error> {
+    let room = client.get_room(&room_id).ok_or_else(Error::room_unknown)?;
+    if room.encryption_state().is_encrypted() && !room.are_members_synced() {
+        room.sync_members()
+            .await
+            .map_err(|source| Error::from_sdk("syncing room members failed", source))?;
+    }
+    Ok(())
+}

Modified file: src/model.rs+3 additions@@ -113,6 +113,9 @@ pub enum RoomCommand {
         room_id: OwnedRoomId,
         typing: bool,
     },
+    /// the key work a send into an encrypted room would otherwise start with, done now.
+    /// nothing in a plain room, and a second ask joins the first
+    PrepareToSend(OwnedRoomId),
     SetAccountData {
         room_id: OwnedRoomId,
         event_type: RoomAccountDataEventType,

Modified file: src/transport.rs+5 additions@@ -360,6 +360,11 @@ impl ClientHandle {
             .await
     }
 
+    pub async fn prepare_to_send(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        self.run(Command::Room(RoomCommand::PrepareToSend(room_id)))
+            .await
+    }
+
     pub async fn join_room(
         &self,
         room: OwnedRoomOrAliasId,

Added file: tests/prepare.rs+74 additions@@ -0,0 +1,74 @@
+use std::time::Duration;
+
+use macltok::{Engine, EngineConfig};
+use matrix_sdk::{
+    ruma::{api::client::sync::sync_events::v5, room_id, user_id},
+    test_utils::mocks::MatrixMockServer,
+};
+use matrix_sdk_test::{JoinedRoomBuilder, event_factory::EventFactory};
+use wiremock::ResponseTemplate;
+
+async fn prepare_twice(encrypted: bool, member_syncs: u64) {
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_sliding_sync()
+        .ok(v5::Response::new("1".to_owned()))
+        .mount()
+        .await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!prepare:localhost");
+    let factory = EventFactory::new().room(room_id);
+    let mut room = JoinedRoomBuilder::new(room_id).add_state_event(
+        factory
+            .member(user_id!("@example:localhost"))
+            .into_raw_sync_state(),
+    );
+    if encrypted {
+        room = room.add_state_event(
+            factory
+                .room_encryption()
+                .sender(user_id!("@example:localhost"))
+                .into_raw_sync_state(),
+        );
+    }
+    server.sync_room(&client, room).await;
+    let member: matrix_sdk::ruma::serde::Raw<
+        matrix_sdk::ruma::events::room::member::RoomMemberEvent,
+    > = factory.member(user_id!("@example:localhost")).into_raw();
+    server
+        .mock_get_members()
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_body_json(serde_json::json!({ "chunk": [member] }))
+                .set_delay(Duration::from_millis(300)),
+        )
+        .expect(member_syncs)
+        .mount()
+        .await;
+    let config = EngineConfig::default()
+        .with_initial_history_limit(0)
+        .with_prepare_to_send(false);
+    let (engine, handle, mut events) = Engine::from_client(client, config).await.unwrap();
+    let running = tokio::spawn(engine.run());
+    let draining = tokio::spawn(async move { while events.recv().await.is_some() {} });
+    let (first, second) = tokio::join!(
+        handle.prepare_to_send(room_id.to_owned()),
+        handle.prepare_to_send(room_id.to_owned()),
+    );
+    first.unwrap();
+    second.unwrap();
+    handle.shutdown().await.unwrap();
+    running.await.unwrap().unwrap();
+    draining.await.unwrap();
+    server.server().verify().await;
+}
+
+#[tokio::test]
+async fn two_prepares_at_once_sync_members_once() {
+    prepare_twice(true, 1).await;
+}
+
+#[tokio::test]
+async fn a_plain_room_needs_no_preparing() {
+    prepare_twice(false, 0).await;
+}

macltok

Commit 38c4caecdb3d

Commit `38c4caecdb3d`