From 2b247a33772ed32a7ae5ba12a5e1b58cd5014a6f Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Sat, 15 Aug 2026 14:14:15 +0200 Subject: [PATCH 1/2] fix: refill the sliding sync member roster when joined_count rises --- src/app/hooks/useRoomMembers.test.tsx | 54 +++++++++++++++++++++++++- src/app/hooks/useRoomMembers.ts | 19 +++++++-- src/client/roomMemberHydration.test.ts | 22 +++++++++++ src/client/roomMemberHydration.ts | 18 +++++---- 4 files changed, 101 insertions(+), 12 deletions(-) diff --git a/src/app/hooks/useRoomMembers.test.tsx b/src/app/hooks/useRoomMembers.test.tsx index a8176f4646..5f20ef1ddb 100644 --- a/src/app/hooks/useRoomMembers.test.tsx +++ b/src/app/hooks/useRoomMembers.test.tsx @@ -1,6 +1,6 @@ import { act, renderHook, waitFor } from '@testing-library/react'; import { describe, expect, it, vi } from 'vitest'; -import { RoomMemberEvent } from '$types/matrix-sdk'; +import { ClientEvent, RoomMemberEvent } from '$types/matrix-sdk'; import type { MatrixClient, MatrixEvent, Room, RoomMember } from '$types/matrix-sdk'; const { hydrateAllRoomMembers } = vi.hoisted(() => ({ @@ -34,6 +34,58 @@ describe('useRoomMembers', () => { expect(hydrateAllRoomMembers).not.toHaveBeenCalled(); }); + it('refills the roster again on later sync responses', async () => { + hydrateAllRoomMembers.mockClear(); + const room = { + roomId: '!room:example.org', + getMembers: () => [] as RoomMember[], + loadMembersIfNeeded: vi.fn<() => Promise>().mockResolvedValue(undefined), + } as unknown as Room; + const handlers = new Map void>(); + const mx = { + getRoom: () => room, + on: vi.fn<(event: string, handler: () => void) => void>((event, handler) => { + handlers.set(event, handler); + }), + removeListener: vi.fn<() => void>(), + } as unknown as MatrixClient; + + renderHook(() => useRoomMembers(mx, room.roomId)); + + await waitFor(() => expect(hydrateAllRoomMembers).toHaveBeenCalledOnce()); + + act(() => handlers.get(ClientEvent.Sync)?.()); + + expect(hydrateAllRoomMembers).toHaveBeenCalledTimes(2); + }); + + it('does not refill the roster on sync after a failed SDK member load', async () => { + hydrateAllRoomMembers.mockClear(); + const room = { + roomId: '!room:example.org', + getMembers: () => [] as RoomMember[], + loadMembersIfNeeded: vi + .fn<() => Promise>() + .mockRejectedValue(new Error('NetworkError')), + } as unknown as Room; + const handlers = new Map void>(); + const mx = { + getRoom: () => room, + on: vi.fn<(event: string, handler: () => void) => void>((event, handler) => { + handlers.set(event, handler); + }), + removeListener: vi.fn<() => void>(), + } as unknown as MatrixClient; + + renderHook(() => useRoomMembers(mx, room.roomId)); + + await waitFor(() => expect(room.loadMembersIfNeeded).toHaveBeenCalledOnce()); + await Promise.resolve(); + act(() => handlers.get(ClientEvent.Sync)?.()); + + expect(hydrateAllRoomMembers).not.toHaveBeenCalled(); + }); + it('keeps member updates flowing while the SDK member load is pending', async () => { let resolveMemberLoad!: () => void; const members: RoomMember[] = []; diff --git a/src/app/hooks/useRoomMembers.ts b/src/app/hooks/useRoomMembers.ts index 65747c9d44..db05016afd 100644 --- a/src/app/hooks/useRoomMembers.ts +++ b/src/app/hooks/useRoomMembers.ts @@ -1,5 +1,5 @@ import type { MatrixClient, MatrixEvent, RoomMember } from '$types/matrix-sdk'; -import { RoomMemberEvent } from '$types/matrix-sdk'; +import { ClientEvent, RoomMemberEvent } from '$types/matrix-sdk'; import { useEffect, useState } from 'react'; import { hydrateAllRoomMembers } from '$client/roomMemberHydration'; @@ -20,15 +20,23 @@ export const useRoomMembers = (mx: MatrixClient, roomId: string, enabled = true) setMembers(room.getMembers()); }; + // A failed SDK member load must not trigger the direct roster fallback: + // classic sync already owns retries. + let refillAllowed = false; + const refillRoster = () => { + if (!room || disposed || !refillAllowed) return; + void hydrateAllRoomMembers(mx, roomId).then(() => updateMemberList()); + }; + if (room) { setMembers(room.getMembers()); // Sliding sync may retain an incomplete member set. Do not let its SDK - // request block incoming membership updates. A failed request must not - // trigger the direct roster fallback: classic sync already owns retries. + // request block incoming membership updates. void room.loadMembersIfNeeded().then( () => { + refillAllowed = true; updateMemberList(); - void hydrateAllRoomMembers(mx, roomId).then(() => updateMemberList()); + refillRoster(); }, () => updateMemberList() ); @@ -36,10 +44,13 @@ export const useRoomMembers = (mx: MatrixClient, roomId: string, enabled = true) mx.on(RoomMemberEvent.Membership, updateMemberList); mx.on(RoomMemberEvent.PowerLevel, updateMemberList); + // joined_count can rise after mount and emits no event of its own. + mx.on(ClientEvent.Sync, refillRoster); return () => { disposed = true; mx.removeListener(RoomMemberEvent.Membership, updateMemberList); mx.removeListener(RoomMemberEvent.PowerLevel, updateMemberList); + mx.removeListener(ClientEvent.Sync, refillRoster); }; }, [enabled, mx, roomId]); diff --git a/src/client/roomMemberHydration.test.ts b/src/client/roomMemberHydration.test.ts index d447e5bae4..ed4ed8740d 100644 --- a/src/client/roomMemberHydration.test.ts +++ b/src/client/roomMemberHydration.test.ts @@ -241,6 +241,28 @@ describe('hydrateAllRoomMembers', () => { expect(members).toHaveBeenCalledTimes(2); }); + it('retries within the TTL when the joined count grows', async () => { + let joinedCount = 3; + const setStateEvents = vi.fn<() => void>(); + const room = { + roomId: ROOM_ID, + getJoinedMembers: () => [{}, {}] as RoomMember[], + getJoinedMemberCount: () => joinedCount, + getMember: () => null, + currentState: { setStateEvents }, + } as unknown as Room; + const members = vi.fn<() => Promise>(() => Promise.resolve({ chunk: [] })); + const mx = { getRoom: () => room, members } as unknown as MatrixClient; + + await hydrateAllRoomMembers(mx, ROOM_ID); + await hydrateAllRoomMembers(mx, ROOM_ID); + expect(members).toHaveBeenCalledTimes(1); + + joinedCount = 20; + await hydrateAllRoomMembers(mx, ROOM_ID); + expect(members).toHaveBeenCalledTimes(2); + }); + it('swallows a failed fetch', async () => { const { mx, members, setStateEvents } = makeBulkFakes(2, 20); members.mockRejectedValueOnce(new Error('403')); diff --git a/src/client/roomMemberHydration.ts b/src/client/roomMemberHydration.ts index ef5493d27a..db4b20b182 100644 --- a/src/client/roomMemberHydration.ts +++ b/src/client/roomMemberHydration.ts @@ -108,24 +108,28 @@ export const hydrateRoomMember = ( // predates most joins keeps a short roster forever. Refill it from the server. const BULK_TTL_MS = 5 * 60_000; const bulkInFlight = new WeakMap>>(); -const bulkAttemptedAt = new WeakMap>(); +type BulkAttempt = { at: number; joinedCount: number }; +const bulkAttempts = new WeakMap>(); export const hydrateAllRoomMembers = (mx: MatrixClient, roomId: string): Promise => { const room = mx.getRoom(roomId); if (!room) return Promise.resolve(); - if (room.getJoinedMembers().length >= room.getJoinedMemberCount()) return Promise.resolve(); + const joinedCount = room.getJoinedMemberCount(); + if (room.getJoinedMembers().length >= joinedCount) return Promise.resolve(); - const attemptedTs = bulkAttemptedAt.get(mx)?.get(roomId); - if (attemptedTs !== undefined && Date.now() - attemptedTs < BULK_TTL_MS) return Promise.resolve(); + // An attempt made against a smaller joined count must not silence the refill. + const attempt = bulkAttempts.get(mx)?.get(roomId); + if (attempt && Date.now() - attempt.at < BULK_TTL_MS && joinedCount <= attempt.joinedCount) + return Promise.resolve(); const pending = bulkInFlight.get(mx) ?? new Map>(); bulkInFlight.set(mx, pending); const existing = pending.get(roomId); if (existing) return existing; - const attempts = bulkAttemptedAt.get(mx) ?? new Map(); - bulkAttemptedAt.set(mx, attempts); - attempts.set(roomId, Date.now()); + const attempts = bulkAttempts.get(mx) ?? new Map(); + bulkAttempts.set(mx, attempts); + attempts.set(roomId, { at: Date.now(), joinedCount }); const request = mx .members(roomId, undefined, KnownMembership.Leave) From 7d4000a231caeef3fb15bb0e7ddfe141d11c6959 Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Sat, 15 Aug 2026 14:29:19 +0200 Subject: [PATCH 2/2] fix(media): serve Android media over loopback, thumbnail timeline images --- src-tauri/src/network/media_protocol.rs | 92 ++++- .../media_protocol/android_loopback.rs | 336 +++++++++++++++--- .../message/content/ImageContent.tsx | 61 +++- 3 files changed, 436 insertions(+), 53 deletions(-) diff --git a/src-tauri/src/network/media_protocol.rs b/src-tauri/src/network/media_protocol.rs index 7b7375872a..1fef94dcb6 100644 --- a/src-tauri/src/network/media_protocol.rs +++ b/src-tauri/src/network/media_protocol.rs @@ -43,7 +43,9 @@ const CACHE_SUBDIR: &str = "sable-media"; // Inactivity deadline between chunks, so a slow but progressing download is not killed. const READ_TIMEOUT: Duration = Duration::from_secs(30); const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); -const MAX_CONCURRENT_THUMBNAIL_REQUESTS: usize = 4; +// Small and multiplexed over one HTTP/2 connection, so a tight cap only serialises the timeline. +const MAX_CONCURRENT_THUMBNAIL_REQUESTS: usize = 12; +// Originals stay capped: more parallelism just splits the same mobile bandwidth. const MAX_CONCURRENT_DOWNLOAD_REQUESTS: usize = 6; // The frontend mounts (and starts requesting media) before it hands us the session, so a request // may arrive first. `` never retries, so waiting beats answering 503. @@ -59,6 +61,16 @@ const MAX_TEMP_CACHE_BYTES: u64 = 2 * 1024 * 1024 * 1024; // 2 GiB type FetchResult = Result<(String, Option>>, PathBuf), StatusCode>; +/// Uninhabited off Android, so the streaming branches compile out. +#[cfg(target_os = "android")] +type FetchProgress = Option>; +#[cfg(not(target_os = "android"))] +type FetchProgress = Option; + +// Published per flushed batch, so a reader never sees bytes still sitting in the write buffer. +#[cfg(target_os = "android")] +const PROGRESS_FLUSH_BYTES: u64 = 64 * 1024; + pub struct MediaSessionState { session_store: SessionStore, encryption: EncryptionStore, @@ -389,8 +401,41 @@ async fn handle_request( let dir = cache_dir(app).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let temp_dir = temp_cache_dir(app).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + // Wry blocks the webview's `shouldInterceptRequest` thread for all of this and drops the + // response after 30s, so redirect before fetching. Range requests seek media already cached. + #[cfg(target_os = "android")] + if !loopback && range.is_none() { + if let Some(server) = &state.loopback { + let (redirect, pending) = server.redirect_pending(&session, &key); + if let Some(pending) = pending { + let app = app.clone(); + let session = session.clone(); + let key = key.clone(); + tauri::async_runtime::spawn(async move { + let state = app.state::(); + let progress = Some(Arc::clone(&pending)); + let stored = + ensure_cached(&state, &session, &key, media_url, dir, temp_dir, &progress) + .await + .ok() + .and_then(|(content_type, in_memory_body, disk_path)| { + // An in-memory body means there is no file for the loopback to open. + in_memory_body + .is_none() + .then_some((disk_path, content_type)) + }); + if let Some(server) = &state.loopback { + server.publish(&session, &key, stored.clone()); + } + pending.resolve(stored); + }); + } + return Ok(redirect); + } + } + let (content_type, in_memory_body, disk_path) = - ensure_cached(&state, &session, &key, media_url, dir, temp_dir).await?; + ensure_cached(&state, &session, &key, media_url, dir, temp_dir, &None).await?; #[cfg(target_os = "android")] if loopback && in_memory_body.is_none() && content_type.starts_with("video/") { @@ -420,6 +465,7 @@ async fn ensure_cached( media_url: Url, dir: PathBuf, temp_dir: PathBuf, + progress: &FetchProgress, ) -> Result<(String, Option>>, PathBuf), StatusCode> { ensure_cached_with_limits( state, @@ -430,6 +476,7 @@ async fn ensure_cached( temp_dir, MAX_CACHE_BYTES, MAX_TEMP_CACHE_BYTES, + progress, ) .await } @@ -444,6 +491,7 @@ async fn ensure_cached_with_limits( temp_dir: PathBuf, max_persistent_cache_bytes: u64, max_temp_cache_bytes: u64, + progress: &FetchProgress, ) -> Result<(String, Option>>, PathBuf), StatusCode> { let body_path = dir.join(key); let content_type_path = dir.join(format!("{key}.ct")); @@ -521,6 +569,7 @@ async fn ensure_cached_with_limits( temp_content_type_path, max_persistent_cache_bytes, max_temp_cache_bytes, + progress, ) .await; @@ -576,6 +625,7 @@ async fn fetch_and_cache( temp_content_type_path: PathBuf, max_persistent_cache_bytes: u64, max_temp_cache_bytes: u64, + progress: &FetchProgress, ) -> Result<(String, Option>>, PathBuf), StatusCode> { let permit = acquire_lane(state, &media_url).await; @@ -647,7 +697,15 @@ async fn fetch_and_cache( // Plaintext media streams to disk, so peak memory is one chunk instead of the whole file. let staging_path = temp_body_path.with_extension("part"); - match stream_to_staging_file(&mut upstream, temp_dir.clone(), staging_path.clone()).await { + match stream_to_staging_file( + &mut upstream, + temp_dir.clone(), + staging_path.clone(), + progress, + &content_type, + ) + .await + { StreamOutcome::Written(size) => { drop(permit); let (target_dir, target_body, target_ct, max_bytes) = @@ -707,6 +765,8 @@ async fn stream_to_staging_file( upstream: &mut tauri_plugin_http::reqwest::Response, temp_dir: PathBuf, staging_path: PathBuf, + progress: &FetchProgress, + content_type: &str, ) -> StreamOutcome { if tokio::fs::create_dir_all(&temp_dir).await.is_err() { return StreamOutcome::Unstorable; @@ -717,6 +777,16 @@ async fn stream_to_staging_file( let mut file = tokio::io::BufWriter::new(file); let mut written: u64 = 0; + #[cfg(target_os = "android")] + let mut published: u64 = 0; + + // Without a length the response cannot be framed, so readers wait for the finished file. + #[cfg(target_os = "android")] + if let (Some(pending), Some(total)) = (progress, upstream.content_length()) { + pending.begin_stream(staging_path.clone(), content_type.to_owned(), total); + } + #[cfg(not(target_os = "android"))] + let _ = (progress, content_type); loop { match upstream.chunk().await { @@ -728,9 +798,24 @@ async fn stream_to_staging_file( break; } written += chunk.len() as u64; + + #[cfg(target_os = "android")] + if let Some(pending) = progress { + if written - published >= PROGRESS_FLUSH_BYTES { + if tokio::io::AsyncWriteExt::flush(&mut file).await.is_err() { + break; + } + published = written; + pending.advance(written); + } + } } Ok(None) => { if tokio::io::AsyncWriteExt::flush(&mut file).await.is_ok() { + #[cfg(target_os = "android")] + if let Some(pending) = progress { + pending.advance(written); + } return StreamOutcome::Written(written); } break; @@ -1042,6 +1127,7 @@ mod tests { temp, 1024 * 1024, 1024 * 1024, + &None, ) .await; fs::remove_dir_all(root).ok(); diff --git a/src-tauri/src/network/media_protocol/android_loopback.rs b/src-tauri/src/network/media_protocol/android_loopback.rs index c769857007..bb978804bd 100644 --- a/src-tauri/src/network/media_protocol/android_loopback.rs +++ b/src-tauri/src/network/media_protocol/android_loopback.rs @@ -4,8 +4,9 @@ use std::{ io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}, net::{TcpListener, TcpStream}, path::PathBuf, - sync::{Arc, RwLock}, + sync::{Arc, Condvar, Mutex, RwLock}, thread, + time::Duration, }; use sha2::{Digest, Sha256}; @@ -13,9 +14,132 @@ use tauri::http::{header, Response, StatusCode}; use super::session::MediaSession; +// Frees the socket if a fetch never completes, instead of holding a per-host connection forever. +const PENDING_WAIT: Duration = Duration::from_secs(120); +const KEEP_ALIVE_IDLE: Duration = Duration::from_secs(30); + pub(super) struct LoopbackMediaServer { origin: String, - routes: Arc>>, + routes: Arc>>, +} + +#[derive(Clone)] +enum Route { + Ready(CachedMedia), + /// Registered before the fetch starts so the custom protocol can redirect immediately. + Pending(Arc), +} + +/// One in-flight fetch, read by loopback connections while the fetch task writes it. +pub(super) struct PendingMedia { + state: Mutex, + changed: Condvar, +} + +#[derive(Default)] +struct PendingState { + stream: Option, + written: u64, + outcome: Option>, +} + +#[derive(Clone)] +struct PendingStream { + path: PathBuf, + content_type: String, + total: u64, +} + +enum PendingStart { + /// Known length, so the staging file can be tailed as it lands. + Stream(PendingStream), + Done(Option), +} + +impl PendingMedia { + fn new() -> Self { + Self { + state: Mutex::new(PendingState::default()), + changed: Condvar::new(), + } + } + + /// Only plaintext media of known length streams; everything else waits for [`Self::resolve`]. + pub(super) fn begin_stream(&self, path: PathBuf, content_type: String, total: u64) { + if let Ok(mut state) = self.state.lock() { + state.stream = Some(PendingStream { + path, + content_type, + total, + }); + } + self.changed.notify_all(); + } + + pub(super) fn advance(&self, written: u64) { + if let Ok(mut state) = self.state.lock() { + state.written = written; + } + self.changed.notify_all(); + } + + pub(super) fn resolve(&self, media: Option<(PathBuf, String)>) { + let media = media.map(|(path, content_type)| CachedMedia { path, content_type }); + if let Ok(mut state) = self.state.lock() { + state.outcome = Some(media); + } + self.changed.notify_all(); + } + + fn wait_start(&self) -> PendingStart { + let Ok(state) = self.state.lock() else { + return PendingStart::Done(None); + }; + let Ok((state, _)) = self + .changed + .wait_timeout_while(state, PENDING_WAIT, |state| { + state.stream.is_none() && state.outcome.is_none() + }) + else { + return PendingStart::Done(None); + }; + match (&state.stream, &state.outcome) { + // A fetch that already finished is served from its final path, not the staging file. + (_, Some(outcome)) => PendingStart::Done(outcome.clone()), + (Some(stream), None) => PendingStart::Stream(stream.clone()), + (None, None) => PendingStart::Done(None), + } + } + + fn wait_start_done(&self) -> Option { + let state = self.state.lock().ok()?; + let (state, _) = self + .changed + .wait_timeout_while(state, PENDING_WAIT, |state| state.outcome.is_none()) + .ok()?; + state.outcome.clone().flatten() + } + + /// Blocks until more than `have` bytes have landed, or the fetch ends. `None` means no more + /// bytes are coming. + fn wait_for(&self, have: u64) -> Option { + let state = self.state.lock().ok()?; + let (state, _) = self + .changed + .wait_timeout_while(state, PENDING_WAIT, |state| { + state.written <= have && state.outcome.is_none() + }) + .ok()?; + if state.written > have { + return Some(state.written); + } + // Resolved without new bytes: a failure leaves the body short of `total`. + state + .outcome + .as_ref() + .and_then(|outcome| outcome.as_ref())?; + None + } } #[derive(Clone)] @@ -28,7 +152,7 @@ impl LoopbackMediaServer { pub(super) fn start() -> std::io::Result { let listener = TcpListener::bind(("127.0.0.1", 0))?; let origin = format!("http://127.0.0.1:{}", listener.local_addr()?.port()); - let routes = Arc::new(RwLock::new(HashMap::::new())); + let routes = Arc::new(RwLock::new(HashMap::::new())); let server_routes = Arc::clone(&routes); thread::Builder::new() .name("sable-media-loopback".into()) @@ -61,12 +185,61 @@ impl LoopbackMediaServer { if let Ok(mut routes) = self.routes.write() { routes.insert( capability.clone(), - CachedMedia { + Route::Ready(CachedMedia { path, content_type: content_type.to_owned(), - }, + }), ); } + self.redirect_to(&capability) + } + + /// Returns a handle to resolve when the fetch finishes, or `None` if one is already running. + pub(super) fn redirect_pending( + &self, + session: &MediaSession, + cache_key: &str, + ) -> (Response>, Option>) { + let capability = capability(session, cache_key); + let pending = match self.routes.write() { + Ok(mut routes) => { + if routes.contains_key(&capability) { + None + } else { + let pending = Arc::new(PendingMedia::new()); + routes.insert(capability.clone(), Route::Pending(Arc::clone(&pending))); + Some(pending) + } + } + Err(_) => None, + }; + (self.redirect_to(&capability), pending) + } + + /// Swaps the pending entry for the finished file so later requests skip straight to it. + pub(super) fn publish( + &self, + session: &MediaSession, + cache_key: &str, + media: Option<(PathBuf, String)>, + ) { + let capability = capability(session, cache_key); + if let Ok(mut routes) = self.routes.write() { + match &media { + Some((path, content_type)) => routes.insert( + capability, + Route::Ready(CachedMedia { + path: path.clone(), + content_type: content_type.clone(), + }), + ), + // A failed fetch must not linger: the next request has to retry it. + None => routes.remove(&capability), + }; + } + } + + fn redirect_to(&self, capability: &str) -> Response> { Response::builder() .status(StatusCode::FOUND) .header(header::LOCATION, format!("{}/{}", self.origin, capability)) @@ -90,89 +263,166 @@ fn capability(session: &MediaSession, cache_key: &str) -> String { .collect() } -fn serve(mut stream: TcpStream, routes: Arc>>) { - let Ok((method, capability, range)) = parse_request(&stream) else { - let _ = write_status(&mut stream, 400, "Bad Request", &[]); +fn serve(stream: TcpStream, routes: Arc>>) { + let _ = stream.set_read_timeout(Some(KEEP_ALIVE_IDLE)); + let Ok(mut writer) = stream.try_clone() else { return; }; + let mut reader = BufReader::new(stream); + while serve_one(&mut reader, &mut writer, &routes) {} +} + +fn serve_one( + reader: &mut BufReader, + writer: &mut TcpStream, + routes: &Arc>>, +) -> bool { + let Ok((method, capability, range)) = parse_request(reader) else { + let _ = write_status(writer, 400, "Bad Request", &[], false); + return false; + }; if method != "GET" && method != "HEAD" { - let _ = write_status(&mut stream, 405, "Method Not Allowed", &[]); - return; + let _ = write_status(writer, 405, "Method Not Allowed", &[], false); + return false; } - let media = routes + let route = routes .read() .ok() .and_then(|routes| routes.get(&capability).cloned()); + let Some(route) = route else { + let _ = write_status(writer, 404, "Not Found", &[], false); + return false; + }; + // Waiting costs a loopback socket rather than a webview thread, with no 30s ceiling over it. + let media = match route { + Route::Ready(media) => Some(media), + Route::Pending(pending) => match pending.wait_start() { + // A range request wants to seek, which only the finished file can satisfy. + PendingStart::Stream(stream) if range.is_none() => { + return serve_stream(writer, &pending, &stream, &method); + } + PendingStart::Stream(_) => pending.wait_start_done(), + PendingStart::Done(media) => media, + }, + }; let Some(media) = media else { - let _ = write_status(&mut stream, 404, "Not Found", &[]); - return; + let _ = write_status(writer, 504, "Gateway Timeout", &[], false); + return false; }; let Ok(mut file) = File::open(media.path) else { - let _ = write_status(&mut stream, 404, "Not Found", &[]); - return; + let _ = write_status(writer, 404, "Not Found", &[], false); + return false; }; let Ok(total) = file.metadata().map(|metadata| metadata.len()) else { - let _ = write_status(&mut stream, 500, "Internal Server Error", &[]); - return; + let _ = write_status(writer, 500, "Internal Server Error", &[], false); + return false; }; let selection = range.as_deref().and_then(|value| parse_range(value, total)); if range.is_some() && selection.is_none() { let _ = write_status( - &mut stream, + writer, 416, "Range Not Satisfiable", &[("Content-Range", format!("bytes */{total}"))], + false, ); - return; + return false; } let (start, end, partial) = selection.unwrap_or((0, total.saturating_sub(1), false)); let length = end.saturating_sub(start) + 1; - let mut headers = vec![ - ("Content-Type", media.content_type), - ("Content-Length", length.to_string()), - ("Accept-Ranges", "bytes".to_owned()), - ( - "Access-Control-Allow-Origin", - "https://tauri.localhost".to_owned(), - ), - ( - "Cache-Control", - "private, max-age=31536000, immutable".to_owned(), - ), - ]; + let mut headers = media_headers(&media.content_type, length); if partial { headers.push(("Content-Range", format!("bytes {start}-{end}/{total}"))); } if write_status( - &mut stream, + writer, if partial { 206 } else { 200 }, if partial { "Partial Content" } else { "OK" }, &headers, + true, ) .is_err() - || method == "HEAD" { - return; + return false; + } + if method == "HEAD" { + return true; } if file.seek(SeekFrom::Start(start)).is_err() { - return; + return false; } let mut left = length; let mut buffer = [0_u8; 64 * 1024]; while left > 0 { let want = left.min(buffer.len() as u64) as usize; let Ok(read) = file.read(&mut buffer[..want]) else { - return; + return false; }; - if read == 0 || stream.write_all(&buffer[..read]).is_err() { - return; + if read == 0 || writer.write_all(&buffer[..read]).is_err() { + return false; } left -= read as u64; } + true } -fn parse_request(stream: &TcpStream) -> Result<(String, String, Option), ()> { - let mut reader = BufReader::new(stream); +/// Writes the body as the fetch lands it, so the image paints progressively. +fn serve_stream( + writer: &mut TcpStream, + pending: &PendingMedia, + stream: &PendingStream, + method: &str, +) -> bool { + let headers = media_headers(&stream.content_type, stream.total); + if write_status(writer, 200, "OK", &headers, true).is_err() { + return false; + } + if method == "HEAD" { + return true; + } + let Ok(mut file) = File::open(&stream.path) else { + return false; + }; + let mut sent: u64 = 0; + let mut buffer = [0_u8; 64 * 1024]; + while sent < stream.total { + // Progress is only published after a flush, so these bytes are readable. + let Some(available) = pending.wait_for(sent) else { + return false; + }; + while sent < available.min(stream.total) { + let want = (available.min(stream.total) - sent).min(buffer.len() as u64) as usize; + let Ok(read) = file.read(&mut buffer[..want]) else { + return false; + }; + if read == 0 || writer.write_all(&buffer[..read]).is_err() { + return false; + } + sent += read as u64; + } + } + true +} + +fn media_headers(content_type: &str, length: u64) -> Vec<(&'static str, String)> { + vec![ + ("Content-Type", content_type.to_owned()), + ("Content-Length", length.to_string()), + ("Accept-Ranges", "bytes".to_owned()), + ( + "Access-Control-Allow-Origin", + "https://tauri.localhost".to_owned(), + ), + ( + "Cache-Control", + "private, max-age=31536000, immutable".to_owned(), + ), + ] +} + +fn parse_request( + reader: &mut BufReader, +) -> Result<(String, String, Option), ()> { let mut request_line = String::new(); reader.read_line(&mut request_line).map_err(|_| ())?; let mut parts = request_line.split_whitespace(); @@ -227,10 +477,12 @@ fn write_status( status: u16, reason: &str, headers: &[(&str, String)], + keep_alive: bool, ) -> std::io::Result<()> { + let connection = if keep_alive { "keep-alive" } else { "close" }; write!( stream, - "HTTP/1.1 {status} {reason}\r\nConnection: close\r\n" + "HTTP/1.1 {status} {reason}\r\nConnection: {connection}\r\n" )?; for (name, value) in headers { write!(stream, "{name}: {value}\r\n")?; diff --git a/src/app/components/message/content/ImageContent.tsx b/src/app/components/message/content/ImageContent.tsx index 86613d8a9c..6861a64faa 100644 --- a/src/app/components/message/content/ImageContent.tsx +++ b/src/app/components/message/content/ImageContent.tsx @@ -80,6 +80,21 @@ export function checkIfGif(url: string, mimetype?: string, body?: string) { ); } +// Matches Element Web's timeline thumbnail budget. +const TIMELINE_THUMBNAIL_WIDTH = 800; +const TIMELINE_THUMBNAIL_HEIGHT = 600; +const THUMBNAIL_MIN_SOURCE_BYTES = 1024 * 1024; + +// Follows Element Web's `getThumbUrl`, except that unknown dimensions keep the original: stickers +// and custom emoji render through this component too and routinely omit `info`. +function wantsThumbnail(info: IImageInfo | undefined, width: number, height: number): boolean { + if (!info?.w || !info.h || !info.size) return false; + if (info.w <= width && info.h <= height) return false; + // At 1x the thumbnail is already full quality for the box; denser screens keep the original + // until the file is big enough that the bytes matter more than the sharpness. + return window.devicePixelRatio === 1 || info.size > THUMBNAIL_MIN_SOURCE_BYTES; +} + type RenderViewerProps = { src: string; alt: string; @@ -167,10 +182,30 @@ export const ImageContent = as<'div', ImageContentProps>( const isGif = checkIfGif(url, info?.mimetype, body); + const [thumbnailFailed, setThumbnailFailed] = useState(false); + // A caller-supplied edge means it already decided it wants a thumbnail of that size. + const explicitEdge = typeof matrixThumbnailMaxEdge === 'number' && matrixThumbnailMaxEdge > 0; + // Synapse rejects non-integer dimensions with a 400. + const thumbWidth = Math.round(explicitEdge ? matrixThumbnailMaxEdge : TIMELINE_THUMBNAIL_WIDTH); + const thumbHeight = Math.round( + explicitEdge ? matrixThumbnailMaxEdge : TIMELINE_THUMBNAIL_HEIGHT + ); + const usesThumbnail = + !encInfo && // the homeserver cannot scale media it cannot decrypt + !isGif && // scaling drops the animation + !url.startsWith('http') && + !thumbnailFailed && + (explicitEdge || wantsThumbnail(info, thumbWidth, thumbHeight)); + const rawMediaUrl = useMemo(() => { if (url.startsWith('http')) return url; + if (usesThumbnail) { + return ( + mxcUrlToHttp(mx, url, useAuthentication, thumbWidth, thumbHeight, 'scale') ?? undefined + ); + } return mxcUrlToHttp(mx, url, useAuthentication) ?? undefined; - }, [mx, url, useAuthentication]); + }, [mx, url, useAuthentication, usesThumbnail, thumbWidth, thumbHeight]); const resolvedMediaUrl = useRenderableMediaUrl(encInfo ? undefined : rawMediaUrl); @@ -205,12 +240,8 @@ export const ImageContent = as<'div', ImageContentProps>( setViewerFullSrc(null); return undefined; } - if ( - typeof matrixThumbnailMaxEdge !== 'number' || - matrixThumbnailMaxEdge <= 0 || - encInfo || - url.startsWith('http') - ) { + // The timeline shows a scaled rendition, so the viewer has to re-fetch the original. + if (!usesThumbnail) { return undefined; } let cancelled = false; @@ -222,13 +253,18 @@ export const ImageContent = as<'div', ImageContentProps>( return () => { cancelled = true; }; - }, [viewer, matrixThumbnailMaxEdge, encInfo, url, mx, useAuthentication]); + }, [viewer, usesThumbnail, url, mx, useAuthentication]); const handleLoad = () => { setLoad(true); }; const handleError = () => { setLoad(false); + // Homeservers 4xx thumbnail requests for media they cannot scale; the original still works. + if (usesThumbnail) { + setThumbnailFailed(true); + return; + } setError(true); }; @@ -255,6 +291,15 @@ export const ImageContent = as<'div', ImageContentProps>( if (autoPlay) loadSrc().catch(() => undefined); }, [autoPlay, loadSrc]); + // Guarded by a ref rather than `loadSrc` identity: `loadSrc` changes on every render when the + // caller passes `info`/`encInfo` inline, which would otherwise re-fetch in a loop. + const fallbackLoadedRef = useRef(false); + useEffect(() => { + if (!thumbnailFailed || fallbackLoadedRef.current) return; + fallbackLoadedRef.current = true; + loadSrc().catch(() => undefined); + }, [thumbnailFailed, loadSrc]); + const imageW = info?.w; const imageH = info?.h; const hasDimensions = typeof imageW === 'number' && typeof imageH === 'number';