macltok

public

Star0

Fork 0

Watch0

83 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `1f875b103b27`

tarball

atagen <boss@atagen.co>2026-10-25 13:25 UTC

7 files changed+148-51

context: announce sessions closed behind the consumer and keep owners' media apart

A session evicted by the 33rd open or by leaving the room vanished silently, so the next page was the first anyone heard of it. Context media ids now carry their session, because they shared the mentions inbox's ids for the same event and the newer registration's content won. Repeated reads no longer grow a bounded scope's order without limit, and an edit that is still encrypted no longer masks an earlier readable one.

Modified file: src/engine/context.rs+104 additions−36 deletions@@ -13,13 +13,14 @@ use serde_json::Value;
 use tokio::sync::{Mutex, broadcast::error::RecvError};
 
 use crate::{
-    Error, ErrorKind, EventContext, EventContextCommand, EventContextPage, EventContextSession,
-    MediaId, Sender, TimelineEvent, TimelineItem,
+    Error, ErrorKind, Event, EventContext, EventContextCloseReason, EventContextCommand,
+    EventContextPage, EventContextSession, MediaId, Sender, TimelineEvent, TimelineItem,
     engine::MediaService,
     normalise::{
         MediaRegistration, MediaScope, apply_profiles, normalise_context, normalise_observed,
         resolve_replies,
     },
+    transport::EventSender,
 };
 
 const MAX_CONTEXT_LIMIT: u16 = 50;
@@ -27,9 +28,9 @@ const MAX_CONTEXT_LIMIT: u16 = 50;
 const MAX_SESSIONS: usize = 32;
 
 /// open event contexts, oldest first, each owning the media its pages registered
-#[derive(Default)]
 pub(crate) struct Sessions {
     open: Mutex<VecDeque<Session>>,
+    events: EventSender,
 }
 
 struct Session {
@@ -38,36 +39,61 @@ struct Session {
     scope: MediaScope,
 }
 
+impl Session {
+    /// media ids carry the session, so two owners of one event never share an entry
+    fn media_prefix(scope: &MediaScope) -> String {
+        format!("{}#", scope.nonce())
+    }
+}
+
 impl Sessions {
+    pub(crate) fn new(events: EventSender) -> Self {
+        Self {
+            open: Mutex::default(),
+            events,
+        }
+    }
+
     async fn open(
         &self,
         media: &MediaService,
         room_id: OwnedRoomId,
+        scope: MediaScope,
         registrations: Vec<MediaRegistration>,
     ) -> String {
-        let scope = media.context_scope();
         let id = format!("context-{}", scope.nonce());
-        let mut open = self.open.lock().await;
-        media.register(&scope, registrations).await;
-        open.push_back(Session {
-            id: id.clone(),
-            room_id,
-            scope,
-        });
-        while open.len() > MAX_SESSIONS {
-            if let Some(oldest) = open.pop_front() {
-                media.release(&oldest.scope).await;
+        let mut evicted = Vec::new();
+        {
+            let mut open = self.open.lock().await;
+            media.register(&scope, registrations).await;
+            open.push_back(Session {
+                id: id.clone(),
+                room_id,
+                scope,
+            });
+            while open.len() > MAX_SESSIONS {
+                if let Some(oldest) = open.pop_front() {
+                    media.release(&oldest.scope).await;
+                    evicted.push(oldest.id);
+                }
             }
         }
+        self.announce(evicted, EventContextCloseReason::Evicted)
+            .await;
         id
     }
 
-    async fn room(&self, id: &str) -> Result<OwnedRoomId, Error> {
+    /// the room and the media prefix the session's pages use
+    async fn find(&self, id: &str) -> Result<(OwnedRoomId, String), Error> {
         let open = self.open.lock().await;
-        let session = open.iter().find(|session| session.id == id);
-        session
-            .map(|session| session.room_id.clone())
-            .ok_or_else(not_open)
+        let session = open
+            .iter()
+            .find(|session| session.id == id)
+            .ok_or_else(not_open)?;
+        Ok((
+            session.room_id.clone(),
+            Session::media_prefix(&session.scope),
+        ))
     }
 
     /// under the lock, so a page that lands after its close registers nothing
@@ -99,19 +125,37 @@ impl Sessions {
     }
 
     async fn close_left(&self, media: &MediaService, client: &Client) {
-        let mut open = self.open.lock().await;
-        let mut kept = VecDeque::with_capacity(open.len());
-        for session in open.drain(..) {
-            let left = client
-                .get_room(&session.room_id)
-                .is_none_or(|room| matches!(room.state(), RoomState::Left | RoomState::Banned));
-            if left {
-                media.release(&session.scope).await;
-            } else {
-                kept.push_back(session);
+        let mut closed = Vec::new();
+        {
+            let mut open = self.open.lock().await;
+            let mut kept = VecDeque::with_capacity(open.len());
+            for session in open.drain(..) {
+                let left = client.get_room(&session.room_id).is_none_or(|room| {
+                    matches!(room.state(), RoomState::Left | RoomState::Banned)
+                });
+                if left {
+                    media.release(&session.scope).await;
+                    closed.push(session.id);
+                } else {
+                    kept.push_back(session);
+                }
+            }
+            *open = kept;
+        }
+        self.announce(closed, EventContextCloseReason::RoomLeft)
+            .await;
+    }
+
+    async fn announce(&self, ids: Vec<String>, reason: EventContextCloseReason) {
+        for context_id in ids {
+            let event = Event::EventContextClosed {
+                context_id,
+                reason: reason.clone(),
+            };
+            if self.events.send(event).await.is_err() {
+                return;
             }
         }
-        *open = kept;
     }
 
     /// the media service is cleared beside this, so nothing is released one by one
@@ -157,15 +201,17 @@ pub(crate) async fn execute(
             before_limit,
             after_limit,
         } => {
-            let (context, registrations, _) = capture(
+            let scope = media.context_scope();
+            let (context, registrations, _) = capture_in(
                 room(&room_id)?,
                 current_user_id,
                 event_id,
                 before_limit,
                 after_limit,
+                &Session::media_prefix(&scope),
             )
             .await?;
-            let context_id = sessions.open(&media, room_id, registrations).await;
+            let context_id = sessions.open(&media, room_id, scope, registrations).await;
             return Ok(crate::CommandOutput::EventContext(EventContextSession {
                 context_id,
                 context,
@@ -199,7 +245,8 @@ pub(crate) async fn execute(
             "context pagination token is empty",
         ));
     }
-    let room = room(&sessions.room(&context_id).await?)?;
+    let (room_id, media_prefix) = sessions.find(&context_id).await?;
+    let room = room(&room_id)?;
     let side = fetch_side(&room, Some(&token), limit, backwards).await?;
     let mut page = Assembly::new(
         &room,
@@ -207,6 +254,7 @@ pub(crate) async fn execute(
         &side.values,
         &side.highlights,
         None,
+        &media_prefix,
     )
     .await;
     let events = page.take(0..side.values.len());
@@ -224,6 +272,17 @@ pub(crate) async fn capture(
     before_limit: u16,
     after_limit: u16,
 ) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {
+    capture_in(room, current_user_id, event_id, before_limit, after_limit, "").await
+}
+
+async fn capture_in(
+    room: Room,
+    current_user_id: Option<OwnedUserId>,
+    event_id: OwnedEventId,
+    before_limit: u16,
+    after_limit: u16,
+    media_prefix: &str,
+) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {
     validate_limit(before_limit)?;
     validate_limit(after_limit)?;
     let raw = room
@@ -264,6 +323,7 @@ pub(crate) async fn capture(
         &values,
         &highlights,
         Some(target_index),
+        media_prefix,
     )
     .await;
     let events_before = page.take(0..target_index);
@@ -357,8 +417,15 @@ impl Assembly {
         values: &[Value],
         highlights: &[bool],
         keep: Option<usize>,
+        media_prefix: &str,
     ) -> Self {
-        let page = normalise_context(values, keep, current_user_id, room.room_id());
+        let page = normalise_context(
+            values,
+            keep,
+            current_user_id,
+            room.room_id(),
+            media_prefix,
+        );
         let mut items = page.items;
         for (index, item) in &mut items {
             if let TimelineItem::Event(event) = item {
@@ -386,7 +453,7 @@ impl Assembly {
         let fetched = join_all(
             missing
                 .into_iter()
-                .map(|event_id| fetch_reply(room, current_user_id, event_id)),
+                .map(|event_id| fetch_reply(room, current_user_id, event_id, media_prefix)),
         )
         .await;
         for (event, registrations) in fetched.into_iter().flatten() {
@@ -433,6 +500,7 @@ async fn fetch_reply(
     room: &Room,
     current_user_id: Option<&ruma::UserId>,
     event_id: OwnedEventId,
+    media_prefix: &str,
 ) -> Option<(TimelineEvent, Vec<MediaRegistration>)> {
     let event = room
         .load_or_fetch_event(&event_id, None)
@@ -440,7 +508,7 @@ async fn fetch_reply(
         .inspect_err(|error| tracing::debug!(%event_id, %error, "fetching a reply target failed"))
         .ok()?;
     let value: Value = serde_json::from_str(event.raw().json().get()).ok()?;
-    let identity = format!("{}\n{event_id}", room.room_id());
+    let identity = format!("{media_prefix}{}\n{event_id}", room.room_id());
     match normalise_observed(&value, current_user_id, &identity)? {
         (TimelineItem::Event(event), media) => Some((*event, media)),
         _ => None,

Modified file: src/engine/media.rs+7 additions−7 deletions@@ -778,14 +778,19 @@ struct ScopeEntries {
 
 impl ScopeEntries {
     fn touch(&mut self, id: &MediaId) {
-        if self.limit.is_none() {
+        let Some(limit) = self.limit else {
             self.stamps.insert(id.clone(), 0);
             return;
-        }
+        };
         let stamp = self.next_stamp;
         self.next_stamp += 1;
         self.stamps.insert(id.clone(), stamp);
         self.order.push_back((id.clone(), stamp));
+        if self.order.len() > limit.saturating_mul(4) {
+            let stamps = &self.stamps;
+            self.order
+                .retain(|(id, stamp)| stamps.get(id) == Some(stamp));
+        }
     }
 
     /// what fell out past the limit
@@ -803,11 +808,6 @@ impl ScopeEntries {
                 dropped.push(id);
             }
         }
-        if self.order.len() > limit.saturating_mul(4) {
-            let stamps = &self.stamps;
-            self.order
-                .retain(|(id, stamp)| stamps.get(id) == Some(stamp));
-        }
         dropped
     }
 }

Modified file: src/engine/mod.rs+1 additions−1 deletions@@ -404,12 +404,12 @@ impl Engine {
         let (notifications, notification_queue) = mpsc::channel(config.command_capacity);
         let (encryption, encryption_queue) = mpsc::channel(config.command_capacity);
         let engine = Self {
+            contexts: Arc::new(context::Sessions::new(port.event_sender())),
             client,
             sync,
             port,
             config,
             media,
-            contexts: Arc::default(),
             tasks: TaskRegistry {
                 tasks: TaskSet::new(),
             },

Modified file: src/lib.rs+1−1

@@ -57,7 +57,7 @@ pub use mirror::{

5757 pub use model::{

5858 AccountCommand, AttachmentEncryption, AttachmentUpload, Change, Command, CommandOutput,

5959 ConnectionState, CreateRoomOptions, EncryptionCommand, EncryptionState, EngineErrorKind,

60- EngineFailure, Event, EventContext, EventContextCommand, EventContextPage, EventContextSession,

+    EngineFailure, Event, EventContext, EventContextCloseReason, EventContextCommand, EventContextPage, EventContextSession,

6161 EventRef, HistoryState, ImagePack, ImagePackSource, ImagePacks, KeyBackupState, LatestEvent,

6262 LifecycleCommand, LinkPreview, LinkPreviewImage, Media, MediaCommand, MediaDescriptor,

6363 MediaFile, MediaId, MediaKind, MediaLocation, MediaSize, MemberFilter, MemberMembership,

Modified file: src/mirror.rs+8 additions−1 deletions@@ -20,7 +20,7 @@ use tokio::sync::broadcast;
 
 use crate::{
     ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
-    HistoryRepairReport, HistoryState, ImagePacks, RecentHistoryState, RoomSummary,
+    EventContextCloseReason, HistoryRepairReport, HistoryState, ImagePacks, RecentHistoryState, RoomSummary,
     TimelineCommand, TimelineDiff, TimelineId, TimelineItem, TypingState, UploadOutcome,
     VerificationState, VerificationTarget, engine::executor::EngineBoxFuture,
     transport::PublicationRegistration,
@@ -145,6 +145,10 @@ pub enum StateChange {
     },
     HistoryRepair(HistoryRepairReport),
     Typing(TypingState),
+    EventContextClosed {
+        context_id: String,
+        reason: EventContextCloseReason,
+    },
     Upload {
         transaction_id: OwnedTransactionId,
         upload: UploadState,
@@ -407,6 +411,9 @@ impl Mirror {
                     })
                 });
             }
+            Event::EventContextClosed { context_id, reason } => {
+                self.publish(|_| Some(StateChange::EventContextClosed { context_id, reason }));
+            }
             Event::TypingChanged(typing) => self.publish(|state| {
                 if typing.users.is_empty() {
                     state.typing.remove(&typing.room_id);

Modified file: src/model.rs+16 additions−1 deletions@@ -38,7 +38,8 @@ pub enum Command {
 #[non_exhaustive]
 pub enum EventContextCommand {
     /// opens a session that owns the media of everything it hands out. at most 32 stay
-    /// open, opening another closes the oldest, and leaving the room closes its sessions
+    /// open, opening another closes the oldest, and leaving the room closes its sessions,
+    /// both told with `Event::EventContextClosed`
     Fetch {
         room_id: OwnedRoomId,
         event_id: OwnedEventId,
@@ -690,6 +691,14 @@ pub struct EventContext {
     pub end_token: Option<String>,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum EventContextCloseReason {
+    /// opening a 33rd session closed the oldest
+    Evicted,
+    RoomLeft,
+}
+
 /// a session's media stay resolvable until it closes, and past `EngineConfig::max_context_media`
 /// its least recently used are dropped
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -930,6 +939,12 @@ pub enum Event {
     TimelineClosed {
         timeline_id: TimelineId,
     },
+    /// an event context closed by something other than its own `Close`. its id answers
+    /// `EventContextNotOpen` from now on
+    EventContextClosed {
+        context_id: String,
+        reason: EventContextCloseReason,
+    },
     /// a room that stops being watched says so with an empty `users`
     TypingChanged(TypingState),
     MediaUploadProgress {

Modified file: src/normalise/context.rs+11 additions−4 deletions@@ -20,12 +20,14 @@ struct Annotation<'a> {
 
 /// raw events in timeline order, shown the way the live room timeline shows them. edits and
 /// reactions fold into their targets within the page, and redactions, unmatched relations
-/// and thread replies go. `keep` is never dropped
+/// and thread replies go. `keep` is never dropped. `media_prefix` keeps one owner's media ids
+/// apart from another's for the same event
 pub(crate) fn normalise_context(
     values: &[Value],
     keep: Option<usize>,
     current_user_id: Option<&UserId>,
     room_id: &RoomId,
+    media_prefix: &str,
 ) -> ContextPage {
     let mut edits: HashMap<&str, Vec<&Value>> = HashMap::new();
     let mut annotations: HashMap<&str, Vec<Annotation<'_>>> = HashMap::new();
@@ -56,7 +58,12 @@ pub(crate) fn normalise_context(
                 }
             }
             (Some("m.reaction"), ..) => {}
-            (_, Some("m.replace"), Some(target)) => edits.entry(target).or_default().push(value),
+            // an edit still encrypted keeps its relation in the clear but has no new content
+            (_, Some("m.replace"), Some(target)) => {
+                if value.pointer("/content/m.new_content").is_some() {
+                    edits.entry(target).or_default().push(value);
+                }
+            }
             (_, Some("m.thread"), _) => {}
             _ => shown.push(index),
         }
@@ -89,7 +96,7 @@ pub(crate) fn normalise_context(
                     .as_object_mut()?
                     .insert("m.replace".to_owned(), edit.clone());
             }
-            let identity = format!("{room_id}\n{event_id}");
+            let identity = format!("{media_prefix}{room_id}\n{event_id}");
             let (mut item, registrations) = normalise_observed(&value, current_user_id, &identity)?;
             media.extend(registrations);
             if let (TimelineItem::Event(event), Some(annotations)) =
@@ -232,7 +239,7 @@ mod tests {
                 "content": {},
             }),
         ];
-        let page = normalise_context(&values, None, Some(user_id!("@a:x")), room_id!("!room:x"));
+        let page = normalise_context(&values, None, Some(user_id!("@a:x")), room_id!("!room:x"), "");
         assert_eq!(page.items.len(), 1);
         let TimelineItem::Event(event) = &page.items[0].1 else {
             panic!("expected an event");

macltok

Commit 1f875b103b27

Commit `1f875b103b27`