Merge pull request #7705 from element-hq/bma/fixLeak

Close the threaded timelines created to create a poll, share a location or send media
2026-10-31 08:15:02 +00:00 · 2026-09-21 10:18:37 +02:00
parent f8fdfcae56 ba2e2a02d1
commit 95d734efc9
6 changed files with 343 additions and 76 deletions
@@ -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<Timeline> {
+    /**
+     * 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 <T> withTimeline(block: suspend (Timeline) -> Result<T>): Result<T> {
        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)
        }
    }
 }
@@ -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 <T> ReceiveTurbine<T>.awaitFirstItem(): T {
        skipItems(1)
        return awaitItem()
@@ -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<PollContent> = runCatchingExceptions {
-        getTimelineProvider()
-            .getOrThrow()
-            .getActiveTimeline()
-            .timelineItems
-            .first()
-            .asSequence()
-            .filterIsInstance<MatrixTimelineItem.Event>()
-            .first { it.eventId == eventId }
-            .event
-            .content as PollContent
+    suspend fun getPoll(eventId: EventId): Result<PollContent> = withTimeline { timeline ->
+        runCatchingExceptions {
+            timeline.timelineItems
+                .first()
+                .asSequence()
+                .filterIsInstance<MatrixTimelineItem.Event>()
+                .first { it.eventId == eventId }
+                .event
+                .content as PollContent
+        }
    }

    suspend fun savePoll(
@@ -59,52 +56,49 @@ class PollRepository(
        answers: List<String>,
        pollKind: PollKind,
        maxSelections: Int,
-    ): Result<Unit> = 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<Unit> = 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<Unit> =
-        getTimelineProvider().flatMap { timelineProvider ->
-            timelineProvider.getActiveTimeline()
-                .redactEvent(
-                    eventOrTransactionId = pollStartId.toEventOrTransactionId(),
-                    reason = null,
-                )
-        }
+    ): Result<Unit> = withTimeline { timeline ->
+        timeline.redactEvent(
+            eventOrTransactionId = pollStartId.toEventOrTransactionId(),
+            reason = null,
+        )
+    }

-    private suspend fun getTimelineProvider(): Result<TimelineProvider> {
+    /**
+     * 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 <T> withTimeline(block: suspend (Timeline) -> Result<T>): Result<T> {
        return when (timelineMode) {
            is Timeline.Mode.Thread -> {
-                val threadedTimelineResult = room.createTimeline(CreateTimelineParams.Threaded(timelineMode.threadRootId))
-                threadedTimelineResult.map { threadedTimeline ->
-                    object : TimelineProvider {
-                        private val flow = MutableStateFlow<Timeline?>(threadedTimeline)
-                        override fun activeTimelineFlow(): StateFlow<Timeline?> = flow
+                room.createTimeline(CreateTimelineParams.Threaded(timelineMode.threadRootId))
+                    .flatMap { threadedTimeline ->
+                        threadedTimeline.use { block(it) }
                    }
-                }
            }
-            else -> Result.success(defaultTimelineProvider)
+            else -> block(defaultTimelineProvider.getActiveTimeline())
        }
    }
 }
@@ -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,
+    )
+}
@@ -100,9 +100,9 @@ class DefaultMediaSender(
        inReplyToEventId: EventId?,
    ): Result<Unit> {
        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<Unit> {
        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<Timeline> {
+    /**
+     * 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 <T> withTimeline(block: suspend (Timeline) -> Result<T>): Result<T> {
        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)
        }
    }

@@ -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,
    )
 }