macltok

public

Star0

Fork 0

Watch0

73 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `da8896938cf0`

tarball

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

12 files changed+438-109

media: give every registration an owner and free it with its owner

The shared 2048 entry LRU could evict media for rows still on screen whenever context paging or the mentions inbox registered enough, and context media were never freed at all. Each MediaScope now owns what it registers, an entry lives until every owning scope releases it, and only bounded scopes trim, and only their own least recently used. Event contexts become sessions with an id, closed explicitly, by leaving the room or by opening a 33rd, and EngineConfig::max_media_sources becomes max_context_media to say what it now bounds.

Modified file: src/engine/context.rs+165 additions−17 deletions@@ -1,43 +1,190 @@
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, VecDeque},
+    sync::Arc,
+};
 
 use futures_util::future::{join_all, try_join};
 use matrix_sdk::{
-    Room, deserialized_responses::TimelineEvent as SdkTimelineEvent, room::MessagesOptions,
+    Client, Room, RoomState, deserialized_responses::TimelineEvent as SdkTimelineEvent,
+    room::MessagesOptions,
 };
-use ruma::{OwnedEventId, OwnedUserId, UInt, push::Action};
+use ruma::{OwnedEventId, OwnedRoomId, OwnedUserId, UInt, push::Action};
 use serde_json::Value;
+use tokio::sync::{Mutex, broadcast::error::RecvError};
 
 use crate::{
-    Error, ErrorKind, EventContext, EventContextCommand, EventContextPage, MediaId, Sender,
-    TimelineEvent, TimelineItem,
+    Error, ErrorKind, EventContext, EventContextCommand, EventContextPage, EventContextSession,
+    MediaId, Sender, TimelineEvent, TimelineItem,
     engine::MediaService,
     normalise::{
-        MediaRegistration, apply_profiles, normalise_context, normalise_observed, resolve_replies,
+        MediaRegistration, MediaScope, apply_profiles, normalise_context, normalise_observed,
+        resolve_replies,
     },
 };
 
 const MAX_CONTEXT_LIMIT: u16 = 50;
+/// a consumer that never closes its jump views still holds at most this many
+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>>,
+}
+
+struct Session {
+    id: String,
+    room_id: OwnedRoomId,
+    scope: MediaScope,
+}
+
+impl Sessions {
+    async fn open(
+        &self,
+        media: &MediaService,
+        room_id: OwnedRoomId,
+        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;
+            }
+        }
+        id
+    }
+
+    async fn room(&self, id: &str) -> Result<OwnedRoomId, 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)
+    }
+
+    /// under the lock, so a page that lands after its close registers nothing
+    async fn register(
+        &self,
+        media: &MediaService,
+        id: &str,
+        registrations: Vec<MediaRegistration>,
+    ) -> Result<(), Error> {
+        let open = self.open.lock().await;
+        let session = open
+            .iter()
+            .find(|session| session.id == id)
+            .ok_or_else(not_open)?;
+        media.register(&session.scope, registrations).await;
+        Ok(())
+    }
+
+    async fn close(&self, media: &MediaService, id: &str) -> Result<(), Error> {
+        let mut open = self.open.lock().await;
+        let index = open
+            .iter()
+            .position(|session| session.id == id)
+            .ok_or_else(not_open)?;
+        if let Some(session) = open.remove(index) {
+            media.release(&session.scope).await;
+        }
+        Ok(())
+    }
+
+    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);
+            }
+        }
+        *open = kept;
+    }
+
+    /// the media service is cleared beside this, so nothing is released one by one
+    pub(crate) async fn clear(&self) {
+        self.open.lock().await.clear();
+    }
+}
+
+fn not_open() -> Error {
+    Error::new(
+        ErrorKind::EventContextNotOpen,
+        "event context is closed or was never opened",
+    )
+}
+
+/// leaving a room closes its event contexts
+pub(super) async fn close_on_leave(
+    client: Client,
+    media: Arc<MediaService>,
+    sessions: Arc<Sessions>,
+) {
+    let mut info = client.room_info_notable_update_receiver();
+    loop {
+        match info.recv().await {
+            Ok(_) | Err(RecvError::Lagged(_)) => sessions.close_left(&media, &client).await,
+            Err(RecvError::Closed) => return,
+        }
+    }
+}
 
 pub(crate) async fn execute(
-    room: Room,
-    current_user_id: Option<OwnedUserId>,
+    client: Client,
     media: Arc<MediaService>,
+    sessions: Arc<Sessions>,
     command: EventContextCommand,
 ) -> Result<crate::CommandOutput, Error> {
-    let (token, limit, backwards) = match command {
+    let current_user_id = client.user_id().map(ToOwned::to_owned);
+    let room = |room_id: &OwnedRoomId| client.get_room(room_id).ok_or_else(Error::room_unknown);
+    let (context_id, token, limit, backwards) = match command {
         EventContextCommand::Fetch {
+            room_id,
             event_id,
             before_limit,
             after_limit,
-            ..
         } => {
-            let (context, registrations, _) =
-                capture(room, current_user_id, event_id, before_limit, after_limit).await?;
-            media.register(registrations).await;
-            return Ok(crate::CommandOutput::EventContext(context));
+            let (context, registrations, _) = capture(
+                room(&room_id)?,
+                current_user_id,
+                event_id,
+                before_limit,
+                after_limit,
+            )
+            .await?;
+            let context_id = sessions.open(&media, room_id, registrations).await;
+            return Ok(crate::CommandOutput::EventContext(EventContextSession {
+                context_id,
+                context,
+            }));
+        }
+        EventContextCommand::Close { context_id } => {
+            sessions.close(&media, &context_id).await?;
+            return Ok(crate::CommandOutput::Done);
         }
-        EventContextCommand::Before { token, limit, .. } => (token, limit, true),
-        EventContextCommand::After { token, limit, .. } => (token, limit, false),
+        EventContextCommand::Before {
+            context_id,
+            token,
+            limit,
+        } => (context_id, token, limit, true),
+        EventContextCommand::After {
+            context_id,
+            token,
+            limit,
+        } => (context_id, token, limit, false),
     };
     if limit == 0 {
         return Err(Error::new(
@@ -52,6 +199,7 @@ pub(crate) async fn execute(
             "context pagination token is empty",
         ));
     }
+    let room = room(&sessions.room(&context_id).await?)?;
     let side = fetch_side(&room, Some(&token), limit, backwards).await?;
     let mut page = Assembly::new(
         &room,
@@ -62,7 +210,7 @@ pub(crate) async fn execute(
     )
     .await;
     let events = page.take(0..side.values.len());
-    media.register(page.media).await;
+    sessions.register(&media, &context_id, page.media).await?;
     Ok(crate::CommandOutput::EventContextPage(EventContextPage {
         events,
         next_token: side.token,

Modified file: src/engine/media.rs+160 additions−39 deletions@@ -56,27 +56,34 @@ pub(crate) struct MediaService {
     registry: Mutex<MediaRegistry>,
     flights: Mutex<HashMap<String, WeakShared<FlightFuture>>>,
     max_bytes: usize,
+    context_limit: usize,
     url_previews: UrlPreviews,
 }
 
 impl MediaService {
-    pub(crate) fn new(client: Client, max_bytes: usize, max_sources: usize) -> Self {
+    pub(crate) fn new(client: Client, max_bytes: usize, context_limit: usize) -> Self {
         Self {
             url_previews: UrlPreviews::new(client.clone()),
             client,
-            registry: Mutex::new(MediaRegistry::new(max_sources)),
+            registry: Mutex::new(MediaRegistry::default()),
             flights: Mutex::new(HashMap::new()),
             max_bytes,
+            context_limit,
         }
     }
 
+    /// one event context session's media, bounded by `EngineConfig::max_context_media`
+    pub(crate) fn context_scope(&self) -> MediaScope {
+        MediaScope::bounded(self.context_limit)
+    }
+
     pub(crate) async fn url_preview(&self, url: &str) -> Result<Option<LinkPreview>, Error> {
         self.url_previews.preview(url).await
     }
 
-    pub(crate) async fn register(&self, media: Vec<MediaRegistration>) {
+    pub(crate) async fn register(&self, scope: &MediaScope, media: Vec<MediaRegistration>) {
         if !media.is_empty() {
-            self.registry.lock().await.insert_all(media);
+            self.registry.lock().await.insert_all(scope, media);
         }
     }
 
@@ -747,64 +754,178 @@ async fn check_reply_thread(
     }
 }
 
-/// least recently seen goes first. stale order slots are skipped by stamp rather than hunted down
+/// every entry belongs to the scopes that registered it and lives until the last of them
+/// lets go. there is no global cap, a bounded scope only ever trims itself
+#[derive(Default)]
 struct MediaRegistry {
-    entries: HashMap<MediaId, (MediaRegistration, u64)>,
+    entries: HashMap<MediaId, Entry>,
+    scopes: HashMap<u64, ScopeEntries>,
+}
+
+struct Entry {
+    media: MediaRegistration,
+    owners: Vec<u64>,
+}
+
+/// least recently used first. stale order slots are skipped by stamp rather than hunted down
+#[derive(Default)]
+struct ScopeEntries {
+    stamps: HashMap<MediaId, u64>,
     order: VecDeque<(MediaId, u64)>,
     next_stamp: u64,
-    limit: usize,
+    limit: Option<usize>,
 }
 
-impl MediaRegistry {
-    fn new(limit: usize) -> Self {
-        Self {
-            entries: HashMap::new(),
-            order: VecDeque::new(),
-            next_stamp: 0,
-            limit,
+impl ScopeEntries {
+    fn touch(&mut self, id: &MediaId) {
+        if self.limit.is_none() {
+            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));
     }
 
-    fn get(&self, id: &MediaId) -> Option<&MediaRegistration> {
-        self.entries.get(id).map(|(media, _)| media)
-    }
-
-    fn insert_all(&mut self, media: Vec<MediaRegistration>) {
-        for media in media {
-            let stamp = self.next_stamp;
-            self.next_stamp += 1;
-            self.order.push_back((media.id.clone(), stamp));
-            self.entries.insert(media.id.clone(), (media, stamp));
-        }
-        while self.entries.len() > self.limit {
+    /// what fell out past the limit
+    fn trim(&mut self) -> Vec<MediaId> {
+        let Some(limit) = self.limit else {
+            return Vec::new();
+        };
+        let mut dropped = Vec::new();
+        while self.stamps.len() > limit {
             let Some((id, stamp)) = self.order.pop_front() else {
                 break;
             };
-            if self.is_current(&id, stamp) {
-                self.entries.remove(&id);
+            if self.stamps.get(&id) == Some(&stamp) {
+                self.stamps.remove(&id);
+                dropped.push(id);
             }
         }
-        if self.order.len() > self.limit.saturating_mul(4) {
-            let entries = &self.entries;
+        if self.order.len() > limit.saturating_mul(4) {
+            let stamps = &self.stamps;
             self.order
-                .retain(|(id, stamp)| entries.get(id).is_some_and(|(_, current)| current == stamp));
+                .retain(|(id, stamp)| stamps.get(id) == Some(stamp));
         }
+        dropped
     }
+}
 
-    fn is_current(&self, id: &MediaId, stamp: u64) -> bool {
-        self.entries
-            .get(id)
-            .is_some_and(|(_, current)| *current == stamp)
+impl MediaRegistry {
+    fn get(&mut self, id: &MediaId) -> Option<&MediaRegistration> {
+        let entry = self.entries.get(id)?;
+        for owner in &entry.owners {
+            if let Some(scope) = self.scopes.get_mut(owner) {
+                scope.touch(id);
+            }
+        }
+        Some(&entry.media)
+    }
+
+    fn insert_all(&mut self, scope: &MediaScope, media: Vec<MediaRegistration>) {
+        let owner = scope.nonce();
+        let entries = self.scopes.entry(owner).or_insert_with(|| ScopeEntries {
+            limit: scope.limit(),
+            ..ScopeEntries::default()
+        });
+        for media in media {
+            let id = media.id.clone();
+            entries.touch(&id);
+            let entry = self.entries.entry(id).or_insert_with(|| Entry {
+                media: media.clone(),
+                owners: Vec::new(),
+            });
+            entry.media = media;
+            if !entry.owners.contains(&owner) {
+                entry.owners.push(owner);
+            }
+        }
+        for id in entries.trim() {
+            Self::disown(&mut self.entries, &id, owner);
+        }
+    }
+
+    fn disown(entries: &mut HashMap<MediaId, Entry>, id: &MediaId, owner: u64) {
+        let Some(entry) = entries.get_mut(id) else {
+            return;
+        };
+        entry.owners.retain(|current| *current != owner);
+        if entry.owners.is_empty() {
+            entries.remove(id);
+        }
     }
 
     fn release(&mut self, scope: &MediaScope) {
-        self.entries.retain(|id, _| !scope.owns(id));
-        let entries = &self.entries;
-        self.order.retain(|(id, _)| entries.contains_key(id));
+        let Some(released) = self.scopes.remove(&scope.nonce()) else {
+            return;
+        };
+        for id in released.stamps.keys() {
+            Self::disown(&mut self.entries, id, scope.nonce());
+        }
     }
 
     fn clear(&mut self) {
         self.entries.clear();
-        self.order.clear();
+        self.scopes.clear();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::MediaKind;
+
+    fn media(name: &str) -> MediaRegistration {
+        let uri = ruma::OwnedMxcUri::from(format!("mxc://example.org/{name}"));
+        MediaRegistration {
+            id: MediaId::new(format!("test#{name}")),
+            kind: MediaKind::Image,
+            full: MediaSourceInfo {
+                source: MediaSource::Plain(uri),
+                mimetype: None,
+                declared_size: None,
+            },
+            thumbnail: None,
+        }
+    }
+
+    fn has(registry: &mut MediaRegistry, name: &str) -> bool {
+        registry.get(&media(name).id).is_some()
+    }
+
+    #[test]
+    fn shared_media_lives_until_every_owner_lets_go() {
+        let mut registry = MediaRegistry::default();
+        let (timeline, context) = (MediaScope::new(), MediaScope::bounded(8));
+        registry.insert_all(&timeline, vec![media("shared"), media("row")]);
+        registry.insert_all(&context, vec![media("shared"), media("page")]);
+        registry.release(&context);
+        assert!(has(&mut registry, "shared"));
+        assert!(has(&mut registry, "row"));
+        assert!(!has(&mut registry, "page"));
+        registry.release(&timeline);
+        assert!(!has(&mut registry, "shared"));
+        assert!(registry.entries.is_empty());
+    }
+
+    #[test]
+    fn heavy_context_paging_trims_only_its_own_least_recent_entries() {
+        let mut registry = MediaRegistry::default();
+        let (timeline, context) = (MediaScope::new(), MediaScope::bounded(4));
+        registry.insert_all(&timeline, vec![media("on-screen"), media("both")]);
+        registry.insert_all(&context, vec![media("both"), media("kept")]);
+        for page in 0..50 {
+            assert!(has(&mut registry, "kept"));
+            registry.insert_all(&context, vec![media(&format!("page-{page}"))]);
+        }
+        assert!(has(&mut registry, "on-screen"));
+        assert!(has(&mut registry, "both"));
+        assert!(has(&mut registry, "kept"));
+        assert!(!has(&mut registry, "page-0"));
+        assert!(registry.scopes[&context.nonce()].stamps.len() <= 4);
+        registry.release(&context);
+        assert!(has(&mut registry, "both"));
+        assert!(!has(&mut registry, "kept"));
     }
 }

Modified file: src/engine/mod.rs+22 additions−19 deletions@@ -64,7 +64,8 @@ pub struct EngineConfig {
     /// what a thread paginates as it opens, and a room if its recent load fails. 0 for neither
     pub initial_history_limit: u16,
     pub max_media_bytes: usize,
-    pub max_media_sources: usize,
+    /// media one event context session keeps registered, least recently used dropped first
+    pub max_context_media: usize,
     pub max_mentions_bytes: usize,
     pub max_mentions_entries: usize,
     pub max_mention_context_events: usize,
@@ -123,8 +124,8 @@ impl EngineConfig {
     }
 
     #[must_use]
-    pub const fn with_max_media_sources(mut self, sources: usize) -> Self {
-        self.max_media_sources = sources;
+    pub const fn with_max_context_media(mut self, media: usize) -> Self {
+        self.max_context_media = media;
         self
     }
 
@@ -142,7 +143,7 @@ impl Default for EngineConfig {
             event_capacity: 128,
             initial_history_limit: 30,
             max_media_bytes: 20 * 1024 * 1024,
-            max_media_sources: 2_048,
+            max_context_media: 1_024,
             max_mentions_bytes: 64 * 1024 * 1024,
             max_mentions_entries: 5_000,
             max_mention_context_events: 25_000,
@@ -302,6 +303,7 @@ pub struct Engine {
     port: DriverPort,
     config: EngineConfig,
     media: Arc<MediaService>,
+    contexts: Arc<context::Sessions>,
     tasks: TaskRegistry,
     rooms: HashMap<OwnedRoomId, RoomSlot>,
     next_generation: u64,
@@ -351,7 +353,7 @@ impl Engine {
         }
         if config.command_capacity == 0
             || config.event_capacity == 0
-            || config.max_media_sources == 0
+            || config.max_context_media == 0
             || config.max_mentions_entries == 0
             || config.max_mention_context_events == 0
             || config.max_mention_candidates == 0
@@ -362,7 +364,7 @@ impl Engine {
         let media = Arc::new(MediaService::new(
             client.clone(),
             config.max_media_bytes,
-            config.max_media_sources,
+            config.max_context_media,
         ));
         let (mentions, mentions_service, mentions_updates) =
             crate::mentions::service::Service::new(
@@ -407,6 +409,7 @@ impl Engine {
             port,
             config,
             media,
+            contexts: Arc::default(),
             tasks: TaskRegistry {
                 tasks: TaskSet::new(),
             },
@@ -535,7 +538,7 @@ impl Engine {
         let client = &self.client;
         let events = || self.port.event_sender();
         let (notices, notice_queue) = mpsc::unbounded_channel();
-        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 13] = [
+        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 14] = [
             (
                 "upload tracker",
                 Box::pin(watchers::uploads(
@@ -600,6 +603,14 @@ impl Engine {
                 Box::pin(watchers::rooms(client.clone(), events())),
             ),
             (
+                "event context watcher",
+                Box::pin(context::close_on_leave(
+                    client.clone(),
+                    self.media.clone(),
+                    self.contexts.clone(),
+                )),
+            ),
+            (
                 "image pack watcher",
                 Box::pin(watchers::image_packs(client.clone(), events())),
             ),
@@ -815,20 +826,11 @@ impl Engine {
             ),
             Command::Timeline(command) => self.route_timeline(command, reply),
             Command::EventContext(command) => {
-                let room_id = match &command {
-                    crate::EventContextCommand::Fetch { room_id, .. }
-                    | crate::EventContextCommand::Before { room_id, .. }
-                    | crate::EventContextCommand::After { room_id, .. } => room_id,
-                };
-                let Some(room) = client.get_room(room_id) else {
-                    reply.done(Err(Error::room_unknown()));
-                    return None;
-                };
-                let current_user_id = client.user_id().map(ToOwned::to_owned);
-                let media = self.media.clone();
+                let (client, media, contexts) =
+                    (client.clone(), self.media.clone(), self.contexts.clone());
                 self.tasks
                     .spawn(TaskContext::Command("event context"), async move {
-                        reply.send(context::execute(room, current_user_id, media, command).await);
+                        reply.send(context::execute(client, media, contexts, command).await);
                     });
             }
             Command::Mentions(command) => {
@@ -1146,6 +1148,7 @@ impl Engine {
         }
         self.tasks.shutdown().await;
         self.rooms.clear();
+        self.contexts.clear().await;
         self.media.clear().await;
         self.port
             .connection_sender()

Modified file: src/engine/timeline.rs+2 additions−2 deletions@@ -417,7 +417,7 @@ async fn start_watcher(
     let current_user_id = context.current_user_id.clone();
     let (normalised, mut hidden) =
         normalise_timeline(items.iter(), current_user_id.as_deref(), scope);
-    context.media.register(normalised.media).await;
+    context.media.register(scope, normalised.media).await;
     context
         .events
         .send(Event::TimelineReset {
@@ -462,7 +462,7 @@ async fn start_watcher(
                     &scope,
                     &mut hidden,
                 );
-                media.register(normalised.media).await;
+                media.register(&scope, normalised.media).await;
                 replies.fetch(normalised.missing_replies);
                 if normalised.diffs.is_empty() {
                     continue;

Modified file: src/error.rs+1

@@ -53,6 +53,7 @@ pub enum ErrorKind {

5353 RoomUnknown,

5454 RoomNotOpen,

5555 TimelineNotOpen,

56+ EventContextNotOpen,

5657 EventNotFound,

5758 RoomBusy,

5859 QueueFull,

Modified file: src/lib.rs+2 additions−2 deletions@@ -57,8 +57,8 @@ pub use mirror::{
 pub use model::{
     AccountCommand, AttachmentEncryption, AttachmentUpload, Change, Command, CommandOutput,
     ConnectionState, CreateRoomOptions, EncryptionCommand, EncryptionState, EngineErrorKind,
-    EngineFailure, Event, EventContext, EventContextCommand, EventContextPage, EventRef,
-    HistoryState, ImagePack, ImagePackSource, ImagePacks, KeyBackupState, LatestEvent,
+    EngineFailure, Event, EventContext, EventContextCommand, EventContextPage, EventContextSession,
+    EventRef, HistoryState, ImagePack, ImagePackSource, ImagePacks, KeyBackupState, LatestEvent,
     LifecycleCommand, LinkPreview, LinkPreviewImage, Media, MediaCommand, MediaDescriptor,
     MediaFile, MediaId, MediaKind, MediaLocation, MediaSize, MemberFilter, MemberMembership,
     MemberSummary, MembershipChange, Message, MessageContent, MessageEditContent, MessageKind,

Modified file: src/mentions/service.rs+11 additions−3 deletions@@ -30,7 +30,7 @@ use crate::{
     MentionGapReason, MentionKey, MentionReason, MentionsCommand, MentionsPage, MentionsUpdate,
     TimelineContent, TimelineItem,
     engine::{MediaService, MentionsStore, context, executor::EngineBoxFuture},
-    normalise::{MediaRegistration, normalise_observed},
+    normalise::{MediaRegistration, MediaScope, normalise_observed},
     transport::EventSender,
 };
 
@@ -60,6 +60,8 @@ pub(crate) struct Service {
     client: Client,
     account: OwnedUserId,
     media: Arc<MediaService>,
+    /// as many media as the inbox keeps events
+    media_scope: MediaScope,
     store: Option<MentionsStore>,
     data: Data,
     sync: SyncState,
@@ -231,8 +233,10 @@ impl Service {
             data.gap(None, *reason, None);
         }
         let page = data.page(&account, None, 50)?;
+        let media_scope = MediaScope::bounded(config.max_mention_context_events);
         media
             .register(
+                &media_scope,
                 page.entries
                     .iter()
                     .filter_map(|entry| data.events.get(&id(&entry.key)))
@@ -247,6 +251,7 @@ impl Service {
             client,
             account,
             media,
+            media_scope,
             store,
             data,
             sync,
@@ -291,6 +296,7 @@ impl Service {
         };
         self.media
             .register(
+                &self.media_scope,
                 page.entries
                     .iter()
                     .filter_map(|entry| self.data.events.get(&id(&entry.key)))
@@ -387,6 +393,7 @@ impl Service {
                 let page = self.data.page(&self.account, cursor.as_deref(), limit)?;
                 self.media
                     .register(
+                        &self.media_scope,
                         page.entries
                             .iter()
                             .filter_map(|entry| self.data.events.get(&id(&entry.key)))
@@ -398,7 +405,7 @@ impl Service {
             }
             MentionsCommand::Context(key) => {
                 let (context, media) = self.data.context(&key)?;
-                self.media.register(media).await;
+                self.media.register(&self.media_scope, media).await;
                 Ok(CommandOutput::MentionContext(context))
             }
             MentionsCommand::RetryContext(key) => {
@@ -429,7 +436,7 @@ impl Service {
                 });
                 self.apply(next).await?;
                 let (context, media) = self.data.context(&key)?;
-                self.media.register(media).await;
+                self.media.register(&self.media_scope, media).await;
                 Ok(CommandOutput::MentionContext(context))
             }
             MentionsCommand::SetRead { keys, read } => {
@@ -1007,6 +1014,7 @@ impl Service {
                         if self.dirty && let Err(error) = self.flush().await {
                             tracing::error!(%error, "saving the mentions inbox at shutdown failed");
                         }
+                        self.media.release(&self.media_scope).await;
                         let _ = reply.send(());
                         return;
                     }

Modified file: src/mentions/service_tests.rs+4 additions−4 deletions@@ -113,7 +113,7 @@ async fn service(client: &Client, store: Option<Result<MentionsStore, Error>>) -
     let media = Arc::new(MediaService::new(
         client.clone(),
         EngineConfig::default().max_media_bytes,
-        EngineConfig::default().max_media_sources,
+        EngineConfig::default().max_context_media,
     ));
     let (_, mut service, _) = Service::new(client.clone(), media, store, EngineConfig::default())
         .await
@@ -412,7 +412,7 @@ async fn service_recovers_notification_history_into_durable_unread_inbox() {
     let media = Arc::new(MediaService::new(
         client.clone(),
         EngineConfig::default().max_media_bytes,
-        EngineConfig::default().max_media_sources,
+        EngineConfig::default().max_context_media,
     ));
     let (handle, service, mut updates) = Service::new(
         client.clone(),
@@ -504,7 +504,7 @@ async fn live_context_survives_capture_then_edit_and_redaction() {
     let media = Arc::new(MediaService::new(
         client.clone(),
         EngineConfig::default().max_media_bytes,
-        EngineConfig::default().max_media_sources,
+        EngineConfig::default().max_context_media,
     ));
     let (_handle, mut service, _updates) =
         Service::new(client, media, None, EngineConfig::default())
@@ -622,7 +622,7 @@ async fn explicit_recovery_rechecks_server_support() {
     let media = Arc::new(MediaService::new(
         client.clone(),
         config.max_media_bytes,
-        config.max_media_sources,
+        config.max_context_media,
     ));
     let (_, mut service, _) = Service::new(client, media, None, config).await.unwrap();
     service.data.coverage.historical_supported = Some(false);

Modified file: src/model.rs+17 additions−3 deletions@@ -37,6 +37,8 @@ pub enum Command {
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[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
     Fetch {
         room_id: OwnedRoomId,
         event_id: OwnedEventId,
@@ -44,15 +46,17 @@ pub enum EventContextCommand {
         after_limit: u16,
     },
     Before {
-        room_id: OwnedRoomId,
+        context_id: String,
         limit: u16,
         token: String,
     },
     After {
-        room_id: OwnedRoomId,
+        context_id: String,
         limit: u16,
         token: String,
     },
+    /// frees the session's media. its rows' media ids stop resolving
+    Close { context_id: String },
 }
 
 #[derive(Debug)]
@@ -668,7 +672,7 @@ pub enum CommandOutput {
         shortcodes: Vec<String>,
         skipped: Vec<String>,
     },
-    EventContext(EventContext),
+    EventContext(EventContextSession),
     EventContextPage(EventContextPage),
     MentionsPage(crate::MentionsPage),
     MentionContext(crate::SavedMentionContext),
@@ -685,6 +689,16 @@ pub struct EventContext {
     pub end_token: Option<String>,
 }
 
+/// 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)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct EventContextSession {
+    pub context_id: String,
+    pub context: EventContext,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]

Modified file: src/normalise/media.rs+20 additions−4 deletions@@ -24,22 +24,38 @@ static NEXT_MEDIA_SCOPE: AtomicU64 = AtomicU64::new(1);
 /// echo replacement, while transaction IDs disappear. Scoping the recycled ID
 /// by a process-wide nonce keeps media IDs stable across replacement without
 /// colliding across rooms, threads, or reopened sessions.
+///
+/// It is also the owner its registrations belong to. Media stays registered while any
+/// scope that registered it is open, and a bounded scope drops its least recently used
+/// entries past its limit without touching anyone else's.
 #[derive(Clone, Debug)]
 pub(crate) struct MediaScope {
     nonce: u64,
+    limit: Option<usize>,
 }
 
 impl MediaScope {
+    /// for a timeline, which holds what is on screen and so is never trimmed
     pub(crate) fn new() -> Self {
         Self {
             nonce: NEXT_MEDIA_SCOPE.fetch_add(1, Ordering::Relaxed),
+            limit: None,
         }
     }
 
-    pub(crate) fn owns(&self, id: &MediaId) -> bool {
-        id.as_str()
-            .split_once('#')
-            .is_some_and(|(nonce, _)| nonce.parse() == Ok(self.nonce))
+    pub(crate) fn bounded(limit: usize) -> Self {
+        Self {
+            limit: Some(limit.max(1)),
+            ..Self::new()
+        }
+    }
+
+    pub(crate) const fn nonce(&self) -> u64 {
+        self.nonce
+    }
+
+    pub(crate) const fn limit(&self) -> Option<usize> {
+        self.limit
     }
 }
 

Modified file: src/transport.rs+23 additions−15 deletions@@ -28,15 +28,15 @@ use tokio::sync::{OnceCell, mpsc, oneshot, watch};
 
 use crate::{
     AccountCommand, AttachmentUpload, Command, CommandOutput, ConnectionState, CreateRoomOptions,
-    EncryptionCommand, EncryptionState, Error, ErrorKind, Event, EventContext, EventContextCommand,
-    EventContextPage, EventRef, HistoryRepairCommand, ImagePackSource, ImagePacks, InteractiveAuth,
-    InteractiveAuthOutcome, LifecycleCommand, LinkPreview, Media, MediaCommand, MediaId,
-    MediaLocation, MediaSize, MemberFilter, MemberSummary, MentionKey, MentionsBoundary,
-    MentionsCommand, MentionsPage, MessageContent, MessageEditContent, NotificationCommand,
-    OwnDevice, PackCommand, PackImageInfo, PackUsage, Patch, PowerLevelChanges, Profile,
-    PushNotification, PushNotificationEvent, PushNotificationStatus, RoomCommand, RoomListFilter,
-    RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomSummary,
-    SavedMentionContext, SpaceChild, TimelineCommand, TimelineId, VerificationAction,
+    EncryptionCommand, EncryptionState, Error, ErrorKind, Event, EventContextCommand,
+    EventContextPage, EventContextSession, EventRef, HistoryRepairCommand, ImagePackSource,
+    ImagePacks, InteractiveAuth, InteractiveAuthOutcome, LifecycleCommand, LinkPreview, Media,
+    MediaCommand, MediaId, MediaLocation, MediaSize, MemberFilter, MemberSummary, MentionKey,
+    MentionsBoundary, MentionsCommand, MentionsPage, MessageContent, MessageEditContent,
+    NotificationCommand, OwnDevice, PackCommand, PackImageInfo, PackUsage, Patch,
+    PowerLevelChanges, Profile, PushNotification, PushNotificationEvent, PushNotificationStatus,
+    RoomCommand, RoomListFilter, RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels,
+    RoomSummary, SavedMentionContext, SpaceChild, TimelineCommand, TimelineId, VerificationAction,
     VerificationTarget,
     encryption::encryption_state,
     engine::{MediaService, StoreLock},
@@ -358,13 +358,14 @@ impl ClientHandle {
         .await
     }
 
+    /// close the session when the view goes, or its media stay registered
     pub async fn event_context(
         &self,
         room_id: &RoomId,
         event_id: &EventId,
         before_limit: u16,
         after_limit: u16,
-    ) -> Result<EventContext, Error> {
+    ) -> Result<EventContextSession, Error> {
         match self
             .execute(Command::EventContext(EventContextCommand::Fetch {
                 room_id: room_id.to_owned(),
@@ -374,19 +375,19 @@ impl ClientHandle {
             }))
             .await?
         {
-            CommandOutput::EventContext(context) => Ok(context),
+            CommandOutput::EventContext(session) => Ok(session),
             output => Err(unexpected("event context", &output)),
         }
     }
 
     pub async fn event_context_before(
         &self,
-        room_id: &RoomId,
+        context_id: &str,
         token: &str,
         limit: u16,
     ) -> Result<EventContextPage, Error> {
         self.event_context_page(EventContextCommand::Before {
-            room_id: room_id.to_owned(),
+            context_id: context_id.to_owned(),
             limit,
             token: token.to_owned(),
         })
@@ -395,18 +396,25 @@ impl ClientHandle {
 
     pub async fn event_context_after(
         &self,
-        room_id: &RoomId,
+        context_id: &str,
         token: &str,
         limit: u16,
     ) -> Result<EventContextPage, Error> {
         self.event_context_page(EventContextCommand::After {
-            room_id: room_id.to_owned(),
+            context_id: context_id.to_owned(),
             limit,
             token: token.to_owned(),
         })
         .await
     }
 
+    pub async fn close_event_context(&self, context_id: &str) -> Result<(), Error> {
+        self.run(Command::EventContext(EventContextCommand::Close {
+            context_id: context_id.to_owned(),
+        }))
+        .await
+    }
+
     async fn event_context_page(
         &self,
         command: EventContextCommand,

Modified file: tests/mentions_inbox.rs+11 additions−1 deletions@@ -717,12 +717,22 @@ async fn event_context_uses_independent_direction_limits_and_saved_mention_conte
         .count();
     assert_eq!(after_saved_request_count, before_saved_request_count);
 
-    let context = handle
+    let session = handle
         .event_context(room_id, context_id, 2, 3)
         .await
         .unwrap();
+    let context = &session.context;
     assert_eq!(context.start_token.as_deref(), Some("manual-before-next"));
     assert_eq!(context.end_token.as_deref(), Some("manual-after-next"));
+    handle
+        .close_event_context(&session.context_id)
+        .await
+        .unwrap();
+    let closed = handle
+        .event_context_before(&session.context_id, "manual-before-next", 5)
+        .await
+        .unwrap_err();
+    assert_eq!(closed.kind, macltok::ErrorKind::EventContextNotOpen);
 
     let requests = server.server().received_requests().await.unwrap();
     let context_limits: Vec<_> = requests

macltok

Commit da8896938cf0

Commit `da8896938cf0`