diff --git a/lib/src/services/memories/memories.service.dart b/lib/src/services/memories/memories.service.dart index 47791c7b..22131068 100644 --- a/lib/src/services/memories/memories.service.dart +++ b/lib/src/services/memories/memories.service.dart @@ -68,6 +68,11 @@ class MemoriesService { static MemoriesState? _cachedState; + /// Media this run has already asked Rust to render a preview for. Every write + /// Rust makes re-fires the stream below, so without this the remaining items + /// would be requested again on every emission. + static final Set _thumbnailRequests = {}; + final _stateController = StreamController.broadcast(); Stream get watchState => _stateController.stream; @@ -84,6 +89,16 @@ class MemoriesService { StreamSubscription>? _dbSubscription; + /// One recompute walks the whole library: every message that references it, + /// every contact, a sort, and a cache write. Every write Rust makes to + /// `media_files` invalidates the query behind it, so a backfill touching a + /// few hundred rows would pay for all of that a few hundred times over on the + /// UI isolate. + static const _minRecomputeGap = Duration(milliseconds: 250); + Timer? _recomputeThrottle; + DateTime? _lastRecomputeAt; + List? _pendingMediaFiles; + /// Instantly pre-warms the gallery state from disk cache during app loading static Future prewarmCache() async { try { @@ -208,7 +223,7 @@ class MemoriesService { await _dbSubscription?.cancel(); _dbSubscription = twonlyDB.mediaFilesDao .watchAllStoredMediaFiles() - .listen(_processMediaFilesStream); + .listen(_onMediaFilesChanged); final pendingFiles = await twonlyDB.mediaFilesDao .getAllMediaFilesPendingMigration(); @@ -317,6 +332,28 @@ class MemoriesService { } } + /// Collapses a burst of row changes into one recompute, always running the + /// most recent batch so nothing is dropped. + void _onMediaFilesChanged(List mediaFiles) { + _pendingMediaFiles = mediaFiles; + if (_recomputeThrottle?.isActive ?? false) return; + + final since = _lastRecomputeAt == null + ? null + : clock.now().difference(_lastRecomputeAt!); + final wait = since == null || since >= _minRecomputeGap + ? Duration.zero + : _minRecomputeGap - since; + + _recomputeThrottle = Timer(wait, () { + final pending = _pendingMediaFiles; + _pendingMediaFiles = null; + if (pending == null) return; + _lastRecomputeAt = clock.now(); + unawaited(_processMediaFilesStream(pending)); + }); + } + Future _processMediaFilesStream(List mediaFiles) async { try { final mediaIds = mediaFiles.map((m) => m.mediaId).toList(); @@ -344,8 +381,10 @@ class MemoriesService { ).copyWith(totalFilesToMigrate: _currentState.totalFilesToMigrate); for (final item in newState.galleryItems) { - if (!item.mediaService.mediaFile.hasThumbnail && - item.mediaService.mediaFile.type != MediaType.audio) { + final mediaFile = item.mediaService.mediaFile; + if (!mediaFile.hasThumbnail && + mediaFile.type != MediaType.audio && + _thumbnailRequests.add(mediaFile.mediaId)) { unawaited(item.mediaService.createThumbnail()); } } @@ -385,6 +424,7 @@ class MemoriesService { } void dispose() { + _recomputeThrottle?.cancel(); _dbSubscription?.cancel(); _stateController.close(); } diff --git a/lib/src/services/memories/memories_cloud.service.dart b/lib/src/services/memories/memories_cloud.service.dart index 6565f785..9a38f8b8 100644 --- a/lib/src/services/memories/memories_cloud.service.dart +++ b/lib/src/services/memories/memories_cloud.service.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:collection'; import 'dart:io'; import 'package:cryptography_flutter_plus/cryptography_flutter_plus.dart'; @@ -56,9 +57,53 @@ class ProgressMultipartRequest extends http.MultipartRequest { } } +/// ChaCha20 for the cloud copies, kept off the platform main thread. +/// +/// The plugin's Android side runs the cipher inline in `onMethodCall`, which +/// Flutter dispatches on the main thread, for every payload from 2 KB up to +/// 20 MB — every thumbnail and every photo. One multi-megabyte media file holds +/// that thread long enough for Android to offer to close the app, and the +/// gallery starts one of these per tile it builds. This runs anything above +/// 10 KB in a background isolate instead, with no upper bound: past the +/// default one a video would fall back to the calling isolate and freeze the +/// UI instead of the platform thread. +Chacha20 _mediaCipher() => BackgroundChacha.poly1305Aead( + channelPolicy: const CryptographyChannelPolicy( + minLength: 10 * 1000, + maxLength: null, + ), +); + class MemoriesCloudService { static final Map _fileMutexes = {}; + /// Cloud transfers allowed to run at once. Every gallery tile without a local + /// file asks for one as it scrolls into view and nothing cancels the request + /// when it scrolls back out, so without a ceiling a fast scroll leaves + /// hundreds of downloads in flight, each holding its whole payload in memory. + static const _maxConcurrentTransfers = 3; + static var _activeTransfers = 0; + static final Queue> _waitingTransfers = Queue(); + + static Future _transferSlot(Future Function() transfer) async { + if (_activeTransfers >= _maxConcurrentTransfers) { + final waiting = Completer(); + _waitingTransfers.add(waiting); + await waiting.future; + } else { + _activeTransfers++; + } + try { + return await transfer(); + } finally { + if (_waitingTransfers.isEmpty) { + _activeTransfers--; + } else { + _waitingTransfers.removeFirst().complete(); + } + } + } + Timer? _timer; bool _isProcessing = false; bool get isProcessing => _isProcessing; @@ -169,29 +214,31 @@ class MemoriesCloudService { return true; } - String? downloadUrl; - try { - downloadUrl = await RustApi.getMemoriesUrl( - mediaId: mediaId, - thumbnail: isThumbnail, - ); - } catch (_) {} - if (downloadUrl == null || downloadUrl.isEmpty) return false; - - try { - final response = await http.get(Uri.parse(downloadUrl)); - - if (response.statusCode == 200) { - return await _decryptFile(response.bodyBytes, targetPath); - } else { - Log.warn( - 'Failed to download ${isThumbnail ? 'thumbnail' : 'full media'} statuscode ${response.statusCode}', + return _transferSlot(() async { + String? downloadUrl; + try { + downloadUrl = await RustApi.getMemoriesUrl( + mediaId: mediaId, + thumbnail: isThumbnail, ); + } catch (_) {} + if (downloadUrl == null || downloadUrl.isEmpty) return false; + + try { + final response = await http.get(Uri.parse(downloadUrl)); + + if (response.statusCode == 200) { + return await _decryptFile(response.bodyBytes, targetPath); + } else { + Log.warn( + 'Failed to download ${isThumbnail ? 'thumbnail' : 'full media'} statuscode ${response.statusCode}', + ); + } + } catch (e) { + Log.warn(e); } - } catch (e) { - Log.warn(e); - } - return false; + return false; + }); }); } @@ -350,7 +397,7 @@ class MemoriesCloudService { String prefix, ) async { final dataToEncrypt = await inputFile.readAsBytes(); - final chacha20 = FlutterChacha20.poly1305Aead(); + final chacha20 = _mediaCipher(); final nonce = chacha20.newNonce(); final secretBox = await chacha20.encrypt( @@ -407,7 +454,7 @@ class MemoriesCloudService { ); // 7. Decrypt the data using the newly retrieved media key - final chacha20 = FlutterChacha20.poly1305Aead(); + final chacha20 = _mediaCipher(); final decryptedBytes = await chacha20.decrypt( secretBox, secretKey: SecretKey(mediaKey), diff --git a/lib/src/visual/views/memories/components/memory_thumbnail.comp.dart b/lib/src/visual/views/memories/components/memory_thumbnail.comp.dart index 84f22c47..0f3fde6e 100644 --- a/lib/src/visual/views/memories/components/memory_thumbnail.comp.dart +++ b/lib/src/visual/views/memories/components/memory_thumbnail.comp.dart @@ -42,6 +42,11 @@ class _MemoriesThumbnailCompState extends State { ImageInfo? _imageInfo; int _retries = 0; bool _hasStoredFile = false; + + /// A tile asks for a missing thumbnail once. Retrying whenever the attempt + /// finishes would spin forever for media whose thumbnail cannot be produced, + /// and every one of those turns is a database round trip on the UI isolate. + bool _recoveryRequested = false; late final ImageStreamListener _listener; @override @@ -69,14 +74,18 @@ class _MemoriesThumbnailCompState extends State { _resolveImage(); } + /// Whether a file is there and has content, in a single stat call. This runs + /// synchronously for every tile the grid builds, so the probe stays cheap. + static bool _isReadable(File file) { + final stat = file.statSync(); + return stat.type != FileSystemEntityType.notFound && stat.size > 0; + } + void _resolveImage() { if (_retries > 3) return; final media = widget.galleryItem.mediaService; - final hasThumbnail = - media.thumbnailPath.existsSync() && - media.thumbnailPath.lengthSync() > 0; - final hasStored = - media.storedPath.existsSync() && media.storedPath.lengthSync() > 0; + final hasThumbnail = _isReadable(media.thumbnailPath); + final hasStored = _isReadable(media.storedPath); _hasStoredFile = hasStored; final isImageOrGif = media.mediaFile.type == MediaType.image || @@ -92,37 +101,41 @@ class _MemoriesThumbnailCompState extends State { _selectedImageFile = media.storedPath; } - if (!hasThumbnail) { - if (hasStored) { - unawaited( - media.createThumbnail().then((_) { - if (mounted) { - _resolveImage(); - } - }), - ); - } else { - unawaited( - MemoriesCloudService.downloadFromCloud(media, isThumbnail: true).then( - (success) { - if (mounted && success) { - _resolveImage(); - } - }, - ), - ); - } + if (!hasThumbnail && !_recoveryRequested) { + _recoveryRequested = true; + // Either source may come back without having written anything, so the + // result is judged by the file itself and the tile falls back to the + // blurhash rather than asking again. + final recovery = hasStored + ? media.createThumbnail() + : MemoriesCloudService.downloadFromCloud(media, isThumbnail: true); + unawaited( + recovery.then((_) { + if (mounted && _isReadable(media.thumbnailPath)) { + _resolveImage(); + } + }), + ); } if (_imageProvider != null) { - _imageStream?.removeListener(_listener); - _imageStream = null; WidgetsBinding.instance.addPostFrameCallback((_) { if (!mounted) return; - final config = createLocalImageConfiguration(context); - _imageStream = _imageProvider?.resolve(config); - _imageStream!.addListener(_listener); + final provider = _imageProvider; + if (provider == null) return; + final stream = provider.resolve(createLocalImageConfiguration(context)); + // Swapping only on a different key keeps the listener count at one: + // dropping the old stream before the new one is resolved would leave + // the listener attached to a completer nothing removes it from, and a + // completer with listeners keeps its decoded image alive. + if (stream.key == _imageStream?.key) return; + _imageStream?.removeListener(_listener); + _imageStream = stream; + stream.addListener(_listener); }); + // A post-frame callback only runs once a frame is coming, and a preview + // that arrived late resolves while the app is sitting idle. + WidgetsBinding.instance.ensureVisualUpdate(); } } @@ -136,6 +149,7 @@ class _MemoriesThumbnailCompState extends State { _imageProvider = null; _imageInfo = null; _retries = 0; + _recoveryRequested = false; _selectedImageFile = null; _resolveImage(); } @@ -205,6 +219,9 @@ class _MemoriesThumbnailCompState extends State { } catch (_) {} if (_retries < 3) { _retries++; + // The broken file is gone, so fetching it again is + // a new attempt rather than the one already made. + _recoveryRequested = false; _resolveImage(); } }); diff --git a/rust/src/services/media_upload.rs b/rust/src/services/media_upload.rs index d8ecd116..1b5e28f6 100644 --- a/rust/src/services/media_upload.rs +++ b/rust/src/services/media_upload.rs @@ -30,7 +30,7 @@ use sqlx::FromRow; use std::collections::HashMap; use std::path::Path; use std::sync::{Arc, LazyLock, Mutex as SyncMutex}; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, Semaphore}; /// A media file with no source bytes and no referencing message is only deleted /// once it is old enough that no in-flight editor can still be holding it. @@ -45,6 +45,13 @@ const RETRY_MARK_GRACE_SECONDS: i64 = 20; static MEDIA_LOCKS: LazyLock>>>> = LazyLock::new(|| SyncMutex::new(HashMap::new())); +/// Backfilling a missing preview decodes a full-resolution photo, so only one +/// runs at a time: a gallery screen asking for every tile it shows would +/// otherwise decode that many photos in parallel. Re-checking the file after +/// the wait also keeps two tiles from writing the same thumbnail. The send path +/// does not take this permit — a capture must not queue behind a backfill. +static THUMBNAIL_BACKFILL: Semaphore = Semaphore::const_new(1); + /// One media file is only ever prepared by one task. Preparation reserves a /// slot, encrypts, and writes request files, so a second concurrent run would /// burn a slot and leave orphaned jobs behind. @@ -719,6 +726,19 @@ impl MediaUploadService { let files = MediaFileService::new(&self.ctx); match kind { "thumbnail" => { + // The caller asks for a thumbnail, not just for the flag to be + // refreshed: a UI that only ever recorded "still missing" would + // ask again on every rebuild and never get a preview. + if !files.thumbnail_path(media_id).exists() { + let permit = THUMBNAIL_BACKFILL.acquire().await; + if !files.thumbnail_path(media_id).exists() { + let Some(media) = self.load(media_id).await? else { + return Ok(()); + }; + self.create_thumbnail(&media).await?; + } + drop(permit); + } self.update_media( "UPDATE media_files SET has_thumbnail = ? WHERE media_id = ?", i64::from(files.thumbnail_path(media_id).exists()), @@ -1430,6 +1450,47 @@ mod tests { assert!(parse_reupload_requested_by(Some("nonsense")).is_empty()); } + /// The gallery asks for a preview through this call and shows a blurhash + /// until the file appears. When it only recorded that the file was still + /// missing, every tile without a preview asked again forever. + #[tokio::test] + async fn a_missing_thumbnail_is_rendered_rather_than_only_recorded() -> anyhow::Result<()> { + let temp = tempfile::tempdir()?; + let data_dir = temp.path().join("data"); + std::fs::create_dir_all(data_dir.join("keyvalue"))?; + std::fs::write( + data_dir.join("keyvalue/user.json"), + serde_json::to_vec(&UserConfig::default())?, + )?; + let ctx = Context::init_for_testing(temp.path().join("database"), data_dir).await?; + + let media_id = "media-thumbnail"; + let database = ctx.app_db.read().await.clone(); + sqlx::query("INSERT INTO media_files(media_id, type, stored) VALUES (?, 'image', 1)") + .bind(media_id) + .execute(&database.pool) + .await?; + drop(database); + + let files = MediaFileService::new(&ctx); + let stored = files.stored_path(media_id, "image"); + MediaFileService::ensure_parent(&stored)?; + image::DynamicImage::ImageRgb8(image::RgbImage::new(900, 1600)).save(&stored)?; + + let service = MediaUploadService::new(&ctx); + service.media_step_finished(media_id, "thumbnail").await?; + + assert!(files.thumbnail_path(media_id).exists()); + let database = ctx.app_db.read().await.clone(); + let has_thumbnail: i64 = + sqlx::query_scalar("SELECT has_thumbnail FROM media_files WHERE media_id = ?") + .bind(media_id) + .fetch_one(&database.pool) + .await?; + assert_eq!(has_thumbnail, 1); + Ok(()) + } + #[test] fn media_lock_is_shared_per_media_id() { let first = media_lock("media-a");