From e2fb2962f618b38d520d73b400b8e0f26d6b7c13 Mon Sep 17 00:00:00 2001 From: Benoit Marty Date: Mon, 14 Sep 2026 15:17:24 +0200 Subject: [PATCH 1/2] Close the threaded timeline created by PollRepository PollRepository created a dedicated timeline for each operation performed in a thread, and never closed it, leaking SDK resources. Replace getTimelineProvider() with a scoped withTimeline() helper, so that the created timeline is closed once the operation completes, whether it succeeds, fails or gets cancelled. The live timeline is left untouched, since it is owned by its room. Co-Authored-By: Claude Opus 5 (1M context) --- .../features/poll/impl/data/PollRepository.kt | 92 +++++----- .../poll/impl/data/PollRepositoryTest.kt | 168 ++++++++++++++++++ 2 files changed, 211 insertions(+), 49 deletions(-) create mode 100644 features/poll/impl/src/test/kotlin/io/element/android/features/poll/impl/data/PollRepositoryTest.kt diff --git a/features/poll/impl/src/main/kotlin/io/element/android/features/poll/impl/data/PollRepository.kt b/features/poll/impl/src/main/kotlin/io/element/android/features/poll/impl/data/PollRepository.kt index d3f177736a..0e98a71d23 100644 --- a/features/poll/impl/src/main/kotlin/io/element/android/features/poll/impl/data/PollRepository.kt +++ b/features/poll/impl/src/main/kotlin/io/element/android/features/poll/impl/data/PollRepository.kt @@ -23,8 +23,6 @@ import io.element.android.libraries.matrix.api.timeline.TimelineProvider import io.element.android.libraries.matrix.api.timeline.getActiveTimeline import io.element.android.libraries.matrix.api.timeline.item.event.PollContent import io.element.android.libraries.matrix.api.timeline.item.event.toEventOrTransactionId -import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.first @AssistedInject @@ -40,17 +38,16 @@ class PollRepository( ): PollRepository } - suspend fun getPoll(eventId: EventId): Result = runCatchingExceptions { - getTimelineProvider() - .getOrThrow() - .getActiveTimeline() - .timelineItems - .first() - .asSequence() - .filterIsInstance() - .first { it.eventId == eventId } - .event - .content as PollContent + suspend fun getPoll(eventId: EventId): Result = withTimeline { timeline -> + runCatchingExceptions { + timeline.timelineItems + .first() + .asSequence() + .filterIsInstance() + .first { it.eventId == eventId } + .event + .content as PollContent + } } suspend fun savePoll( @@ -59,52 +56,49 @@ class PollRepository( answers: List, pollKind: PollKind, maxSelections: Int, - ): Result = when (existingPollId) { - null -> getTimelineProvider().flatMap { timelineProvider -> - timelineProvider - .getActiveTimeline() - .createPoll( - question = question, - answers = answers, - maxSelections = maxSelections, - pollKind = pollKind, - ) - } - else -> getTimelineProvider().flatMap { timelineProvider -> - timelineProvider.getActiveTimeline() - .editPoll( - pollStartId = existingPollId, - question = question, - answers = answers, - maxSelections = maxSelections, - pollKind = pollKind, - ) + ): Result = withTimeline { timeline -> + when (existingPollId) { + null -> timeline.createPoll( + question = question, + answers = answers, + maxSelections = maxSelections, + pollKind = pollKind, + ) + else -> timeline.editPoll( + pollStartId = existingPollId, + question = question, + answers = answers, + maxSelections = maxSelections, + pollKind = pollKind, + ) } } suspend fun deletePoll( pollStartId: EventId, - ): Result = - getTimelineProvider().flatMap { timelineProvider -> - timelineProvider.getActiveTimeline() - .redactEvent( - eventOrTransactionId = pollStartId.toEventOrTransactionId(), - reason = null, - ) - } + ): Result = withTimeline { timeline -> + timeline.redactEvent( + eventOrTransactionId = pollStartId.toEventOrTransactionId(), + reason = null, + ) + } - private suspend fun getTimelineProvider(): Result { + /** + * Invokes [block] with the [Timeline] matching [timelineMode]. + * + * For a thread, a dedicated timeline is created and closed once [block] returns, since such a + * timeline owns SDK resources. For the other modes, the active timeline of + * [defaultTimelineProvider] is used, and is not closed here since it is owned elsewhere. + */ + private suspend fun withTimeline(block: suspend (Timeline) -> Result): Result { return when (timelineMode) { is Timeline.Mode.Thread -> { - val threadedTimelineResult = room.createTimeline(CreateTimelineParams.Threaded(timelineMode.threadRootId)) - threadedTimelineResult.map { threadedTimeline -> - object : TimelineProvider { - private val flow = MutableStateFlow(threadedTimeline) - override fun activeTimelineFlow(): StateFlow = flow + room.createTimeline(CreateTimelineParams.Threaded(timelineMode.threadRootId)) + .flatMap { threadedTimeline -> + threadedTimeline.use { block(it) } } - } } - else -> Result.success(defaultTimelineProvider) + else -> block(defaultTimelineProvider.getActiveTimeline()) } } } diff --git a/features/poll/impl/src/test/kotlin/io/element/android/features/poll/impl/data/PollRepositoryTest.kt b/features/poll/impl/src/test/kotlin/io/element/android/features/poll/impl/data/PollRepositoryTest.kt new file mode 100644 index 0000000000..519643d008 --- /dev/null +++ b/features/poll/impl/src/test/kotlin/io/element/android/features/poll/impl/data/PollRepositoryTest.kt @@ -0,0 +1,168 @@ +/* + * Copyright (c) 2026 Element Creations Ltd. + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial. + * Please see LICENSE files in the repository root for full details. + */ + +package io.element.android.features.poll.impl.data + +import com.google.common.truth.Truth.assertThat +import io.element.android.features.poll.impl.aPollTimelineItems +import io.element.android.features.poll.impl.anOngoingPollContent +import io.element.android.libraries.matrix.api.poll.PollKind +import io.element.android.libraries.matrix.api.room.CreateTimelineParams +import io.element.android.libraries.matrix.api.timeline.Timeline +import io.element.android.libraries.matrix.test.AN_EVENT_ID +import io.element.android.libraries.matrix.test.A_THREAD_ID +import io.element.android.libraries.matrix.test.room.FakeJoinedRoom +import io.element.android.libraries.matrix.test.timeline.FakeTimeline +import io.element.android.libraries.matrix.test.timeline.LiveTimelineProvider +import kotlinx.coroutines.test.runTest +import org.junit.Test + +class PollRepositoryTest { + private val pollContent = anOngoingPollContent() + + @Test + fun `given a thread, getPoll closes the created timeline`() = runTest { + val threadedTimeline = aThreadedTimeline() + val room = aRoomCreating(threadedTimeline) + val sut = createPollRepository(room, Timeline.Mode.Thread(A_THREAD_ID)) + + val result = sut.getPoll(AN_EVENT_ID) + + assertThat(result.getOrNull()).isEqualTo(pollContent) + assertThat(threadedTimeline.closeCounter).isEqualTo(1) + assertThat((room.liveTimeline as FakeTimeline).closeCounter).isEqualTo(0) + } + + @Test + fun `given a thread, savePoll closes the created timeline`() = runTest { + val threadedTimeline = aThreadedTimeline().apply { + createPollLambda = { _, _, _, _ -> Result.success(Unit) } + } + val room = aRoomCreating(threadedTimeline) + val sut = createPollRepository(room, Timeline.Mode.Thread(A_THREAD_ID)) + + val result = sut.savePoll( + existingPollId = null, + question = "A question", + answers = listOf("Yes", "No"), + pollKind = PollKind.Disclosed, + maxSelections = 1, + ) + + assertThat(result.isSuccess).isTrue() + assertThat(threadedTimeline.closeCounter).isEqualTo(1) + } + + @Test + fun `given a thread, savePoll of an existing poll closes the created timeline`() = runTest { + val threadedTimeline = aThreadedTimeline().apply { + editPollLambda = { _, _, _, _, _ -> Result.success(Unit) } + } + val room = aRoomCreating(threadedTimeline) + val sut = createPollRepository(room, Timeline.Mode.Thread(A_THREAD_ID)) + + val result = sut.savePoll( + existingPollId = AN_EVENT_ID, + question = "A question", + answers = listOf("Yes", "No"), + pollKind = PollKind.Disclosed, + maxSelections = 1, + ) + + assertThat(result.isSuccess).isTrue() + assertThat(threadedTimeline.closeCounter).isEqualTo(1) + } + + @Test + fun `given a thread, deletePoll closes the created timeline`() = runTest { + val threadedTimeline = aThreadedTimeline().apply { + redactEventLambda = { _, _ -> Result.success(Unit) } + } + val room = aRoomCreating(threadedTimeline) + val sut = createPollRepository(room, Timeline.Mode.Thread(A_THREAD_ID)) + + val result = sut.deletePoll(AN_EVENT_ID) + + assertThat(result.isSuccess).isTrue() + assertThat(threadedTimeline.closeCounter).isEqualTo(1) + } + + @Test + fun `given a thread, the created timeline is closed even if the operation fails`() = runTest { + val error = RuntimeException("Failed to send the poll") + val threadedTimeline = aThreadedTimeline().apply { + createPollLambda = { _, _, _, _ -> Result.failure(error) } + } + val room = aRoomCreating(threadedTimeline) + val sut = createPollRepository(room, Timeline.Mode.Thread(A_THREAD_ID)) + + val result = sut.savePoll( + existingPollId = null, + question = "A question", + answers = listOf("Yes", "No"), + pollKind = PollKind.Disclosed, + maxSelections = 1, + ) + + assertThat(result.exceptionOrNull()).isEqualTo(error) + assertThat(threadedTimeline.closeCounter).isEqualTo(1) + } + + @Test + fun `given a thread, if the timeline cannot be created the error is returned`() = runTest { + val error = RuntimeException("Failed to create the timeline") + val room = FakeJoinedRoom( + liveTimeline = aLiveTimeline(), + createTimelineResult = { Result.failure(error) }, + ) + val sut = createPollRepository(room, Timeline.Mode.Thread(A_THREAD_ID)) + + val result = sut.deletePoll(AN_EVENT_ID) + + assertThat(result.exceptionOrNull()).isEqualTo(error) + } + + @Test + fun `given a live timeline, the timeline is not closed`() = runTest { + val liveTimeline = aLiveTimeline().apply { + redactEventLambda = { _, _ -> Result.success(Unit) } + } + val room = FakeJoinedRoom(liveTimeline = liveTimeline) + val sut = createPollRepository(room, Timeline.Mode.Live) + + val result = sut.deletePoll(AN_EVENT_ID) + + assertThat(result.isSuccess).isTrue() + assertThat(liveTimeline.closeCounter).isEqualTo(0) + } + + private fun aLiveTimeline() = FakeTimeline( + timelineItems = aPollTimelineItems(mapOf(AN_EVENT_ID to pollContent)), + ) + + private fun aThreadedTimeline() = FakeTimeline( + timelineItems = aPollTimelineItems(mapOf(AN_EVENT_ID to pollContent)), + mode = Timeline.Mode.Thread(A_THREAD_ID), + ) + + private fun aRoomCreating(threadedTimeline: FakeTimeline) = FakeJoinedRoom( + liveTimeline = aLiveTimeline(), + createTimelineResult = { params -> + assertThat(params).isEqualTo(CreateTimelineParams.Threaded(A_THREAD_ID)) + Result.success(threadedTimeline) + }, + ) + + private fun createPollRepository( + room: FakeJoinedRoom, + timelineMode: Timeline.Mode, + ) = PollRepository( + room = room, + defaultTimelineProvider = LiveTimelineProvider(room), + timelineMode = timelineMode, + ) +} From ed250320b3f0e74751d98c95f1e9c7606ea635ba Mon Sep 17 00:00:00 2001 From: Benoit Marty Date: Mon, 14 Sep 2026 15:21:14 +0200 Subject: [PATCH 2/2] Close the threaded timelines created by ShareLocationPresenter and DefaultMediaSender Both created a dedicated timeline for each operation performed in a thread, and never closed it, leaking SDK resources. Apply the same treatment as PollRepository: replace getTimeline() with a scoped withTimeline() helper, which closes the created timeline once the operation completes, whether it succeeds, fails or gets cancelled. In DefaultMediaSender.sendGallery(), waiting for the upload handler is now done inside the block, so that the timeline stays open until the upload completes. Co-Authored-By: Claude Opus 5 (1M context) --- .../impl/share/ShareLocationPresenter.kt | 22 +++++-- .../impl/share/ShareLocationPresenterTest.kt | 39 +++++++++++++ .../mediaupload/impl/DefaultMediaSender.kt | 57 ++++++++++++------- .../impl/DefaultMediaSenderTest.kt | 41 ++++++++++++- 4 files changed, 132 insertions(+), 27 deletions(-) diff --git a/features/location/impl/src/main/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenter.kt b/features/location/impl/src/main/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenter.kt index ccdf598e93..05af96fe5e 100644 --- a/features/location/impl/src/main/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenter.kt +++ b/features/location/impl/src/main/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenter.kt @@ -217,8 +217,8 @@ class ShareLocationPresenter( val replyMode = messageComposerContext.composerMode as? MessageComposerMode.Reply val inReplyToEventId = replyMode?.eventId val geoUri = event.location.toGeoUri() - getTimeline().flatMap { - it.sendLocation( + withTimeline { timeline -> + timeline.sendLocation( body = generateBody(geoUri), geoUri = geoUri, description = null, @@ -237,10 +237,22 @@ class ShareLocationPresenter( ) } - private suspend fun getTimeline(): Result { + /** + * Invokes [block] with the [Timeline] matching [timelineMode]. + * + * For a thread, a dedicated timeline is created and closed once [block] returns, since such a + * timeline owns SDK resources. For the other modes, the live timeline is used, and is not + * closed here since it is owned by its room. + */ + private suspend fun withTimeline(block: suspend (Timeline) -> Result): Result { return when (timelineMode) { - is Timeline.Mode.Thread -> room.createTimeline(CreateTimelineParams.Threaded(timelineMode.threadRootId)) - else -> Result.success(room.liveTimeline) + is Timeline.Mode.Thread -> { + room.createTimeline(CreateTimelineParams.Threaded(timelineMode.threadRootId)) + .flatMap { threadedTimeline -> + threadedTimeline.use { block(it) } + } + } + else -> block(room.liveTimeline) } } } diff --git a/features/location/impl/src/test/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenterTest.kt b/features/location/impl/src/test/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenterTest.kt index 7500d29616..129ea1e4cd 100644 --- a/features/location/impl/src/test/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenterTest.kt +++ b/features/location/impl/src/test/kotlin/io/element/android/features/location/impl/share/ShareLocationPresenterTest.kt @@ -708,6 +708,45 @@ class ShareLocationPresenterTest { } } + @Test + fun `ShareStaticLocation in a thread closes the created timeline`() = runTest { + val threadedTimeline = FakeTimeline(mode = Timeline.Mode.Thread(A_THREAD_ID)).apply { + sendLocationLambda = { _, _, _, _, _, _ -> Result.success(Unit) } + } + val liveTimeline = FakeTimeline() + val joinedRoom = FakeJoinedRoom( + liveTimeline = liveTimeline, + createTimelineResult = { Result.success(threadedTimeline) }, + ) + val shareLocationPresenter = createShareLocationPresenter( + joinedRoom = joinedRoom, + timelineMode = Timeline.Mode.Thread(A_THREAD_ID), + ) + fakePermissionsPresenter.givenState( + aPermissionsState( + permissions = PermissionsState.Permissions.AllGranted, + shouldShowRationale = false, + ) + ) + + shareLocationPresenter.test { + val initialState = awaitFirstItem() + + initialState.eventSink( + ShareLocationEvent.ShareStaticLocation( + location = Location(lat = 3.0, lon = 4.0, accuracy = 5.0f), + isPinned = false, + ) + ) + + advanceUntilIdle() + + assertThat(threadedTimeline.closeCounter).isEqualTo(1) + assertThat(liveTimeline.closeCounter).isEqualTo(0) + cancelAndIgnoreRemainingEvents() + } + } + private suspend fun ReceiveTurbine.awaitFirstItem(): T { skipItems(1) return awaitItem() diff --git a/libraries/mediaupload/impl/src/main/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSender.kt b/libraries/mediaupload/impl/src/main/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSender.kt index 37f77d11f8..36c3ba99fb 100644 --- a/libraries/mediaupload/impl/src/main/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSender.kt +++ b/libraries/mediaupload/impl/src/main/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSender.kt @@ -100,9 +100,9 @@ class DefaultMediaSender( inReplyToEventId: EventId?, ): Result { val mediaLogId = mediaId(mediaUploadInfo.file) - return getTimeline().flatMap { - Timber.d("Started sending media $mediaLogId using timeline: ${it.mode}") - it.sendMedia( + return withTimeline { timeline -> + Timber.d("Started sending media $mediaLogId using timeline: ${timeline.mode}") + timeline.sendMedia( uploadInfo = mediaUploadInfo, caption = caption, formattedCaption = formattedCaption, @@ -128,12 +128,14 @@ class DefaultMediaSender( mediaOptimizationConfig = mediaOptimizationConfig, ) .flatMapCatching { info -> - getTimeline().getOrThrow().sendMedia( - uploadInfo = info, - caption = caption, - formattedCaption = formattedCaption, - inReplyToEventId = inReplyToEventId, - ) + withTimeline { timeline -> + timeline.sendMedia( + uploadInfo = info, + caption = caption, + formattedCaption = formattedCaption, + inReplyToEventId = inReplyToEventId, + ) + } } .handleSendResult(mediaId(uri)) } @@ -158,12 +160,14 @@ class DefaultMediaSender( audioInfo = audioInfo, waveform = waveForm, ) - getTimeline().getOrThrow().sendMedia( - uploadInfo = newInfo, - caption = null, - formattedCaption = null, - inReplyToEventId = inReplyToEventId, - ) + withTimeline { timeline -> + timeline.sendMedia( + uploadInfo = newInfo, + caption = null, + formattedCaption = null, + inReplyToEventId = inReplyToEventId, + ) + } } .handleSendResult(mediaId(uri)) } @@ -176,7 +180,7 @@ class DefaultMediaSender( ): Result { val galleryLogId = "gallery[${mediaUploadInfos.size} items]" Timber.d("Sending $galleryLogId") - return getTimeline().flatMap { timeline -> + return withTimeline { timeline -> val galleryItems = mediaUploadInfos.map { it.toGalleryItemInfo() } timeline.sendGallery( items = galleryItems, @@ -184,10 +188,10 @@ class DefaultMediaSender( formattedCaption = formattedCaption, inReplyToEventId = inReplyToEventId, ) + .flatMapCatching { uploadHandler -> + uploadHandler.await() + } } - .flatMapCatching { uploadHandler -> - uploadHandler.await() - } .handleSendResult(galleryLogId) } @@ -269,12 +273,23 @@ class DefaultMediaSender( } } - private suspend fun getTimeline(): Result { + /** + * Invokes [block] with the [Timeline] matching [timelineMode]. + * + * For a thread, a dedicated timeline is created and closed once [block] returns, since such a + * timeline owns SDK resources. [block] must then perform the whole sending, including waiting + * for the media upload to complete. For the other modes, the live timeline is used, and is not + * closed here since it is owned by its room. + */ + private suspend fun withTimeline(block: suspend (Timeline) -> Result): Result { return when (timelineMode) { is Timeline.Mode.Thread -> { room.createTimeline(CreateTimelineParams.Threaded(threadRootEventId = timelineMode.threadRootId)) + .flatMap { threadedTimeline -> + threadedTimeline.use { block(it) } + } } - else -> Result.success(room.liveTimeline) + else -> block(room.liveTimeline) } } diff --git a/libraries/mediaupload/impl/src/test/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSenderTest.kt b/libraries/mediaupload/impl/src/test/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSenderTest.kt index cfa3585629..97a9b95c51 100644 --- a/libraries/mediaupload/impl/src/test/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSenderTest.kt +++ b/libraries/mediaupload/impl/src/test/kotlin/io/element/android/libraries/mediaupload/impl/DefaultMediaSenderTest.kt @@ -16,6 +16,7 @@ import io.element.android.libraries.matrix.api.media.FileInfo import io.element.android.libraries.matrix.api.media.ImageInfo import io.element.android.libraries.matrix.api.room.JoinedRoom import io.element.android.libraries.matrix.api.timeline.Timeline +import io.element.android.libraries.matrix.test.A_THREAD_ID import io.element.android.libraries.matrix.test.media.FakeMediaUploadHandler import io.element.android.libraries.matrix.test.room.FakeJoinedRoom import io.element.android.libraries.matrix.test.timeline.FakeTimeline @@ -156,14 +157,52 @@ class DefaultMediaSenderTest : RobolectricTest() { sendFileResult.assertions().isCalledOnce() } + @Test + fun `given a thread, the created timeline is closed once the media is sent`() = runTest { + val threadedTimeline = FakeTimeline(mode = Timeline.Mode.Thread(A_THREAD_ID)).apply { + sendFileLambda = { _, _, _, _, _ -> Result.success(FakeMediaUploadHandler()) } + } + val liveTimeline = FakeTimeline() + val room = FakeJoinedRoom( + liveTimeline = liveTimeline, + createTimelineResult = { Result.success(threadedTimeline) }, + ) + val sender = createDefaultMediaSender( + room = room, + timelineMode = Timeline.Mode.Thread(A_THREAD_ID), + ) + + val uri = Uri.parse("content://image.jpg") + val result = sender.sendMedia(uri = uri, mimeType = MimeTypes.Jpeg, mediaOptimizationConfig = mediaOptimizationConfig) + + assertThat(result.isSuccess).isTrue() + assertThat(threadedTimeline.closeCounter).isEqualTo(1) + assertThat(liveTimeline.closeCounter).isEqualTo(0) + } + + @Test + fun `given a live timeline, the timeline is not closed once the media is sent`() = runTest { + val liveTimeline = FakeTimeline().apply { + sendFileLambda = { _, _, _, _, _ -> Result.success(FakeMediaUploadHandler()) } + } + val sender = createDefaultMediaSender(room = FakeJoinedRoom(liveTimeline = liveTimeline)) + + val uri = Uri.parse("content://image.jpg") + val result = sender.sendMedia(uri = uri, mimeType = MimeTypes.Jpeg, mediaOptimizationConfig = mediaOptimizationConfig) + + assertThat(result.isSuccess).isTrue() + assertThat(liveTimeline.closeCounter).isEqualTo(0) + } + private fun createDefaultMediaSender( preProcessor: MediaPreProcessor = FakeMediaPreProcessor(), room: JoinedRoom = FakeJoinedRoom(), + timelineMode: Timeline.Mode = Timeline.Mode.Live, mediaOptimizationConfigProvider: MediaOptimizationConfigProvider = MediaOptimizationConfigProvider { mediaOptimizationConfig }, ) = DefaultMediaSender( preProcessor = preProcessor, room = room, - timelineMode = Timeline.Mode.Live, + timelineMode = timelineMode, mediaOptimizationConfigProvider = mediaOptimizationConfigProvider, ) }