Merge pull request #7184 from element-hq/feature/fga/lls_fix_timeout

Live location sharing timeout
2026-07-16 09:33:02 +02:00
parent e79181d145 a6b5eb4dce
commit 56cf00560e
2 changed files with 72 additions and 9 deletions
@@ -52,9 +52,11 @@ class DefaultActiveLiveLocationShareManager(
    private val clock: SystemClock,
    private val sessionObserver: SessionObserver,
 ) : ActiveLiveLocationShareManager, LiveLocationReceiver {
+    private data class Timeout(val expiresAt: Instant, val job: Job)
+
    private val isSetup = AtomicBoolean(false)
    private val cachedRooms = ConcurrentHashMap<RoomId, JoinedRoom>()
-    private val timeoutJobs = ConcurrentHashMap<RoomId, Job>()
+    private val timeouts = ConcurrentHashMap<RoomId, Timeout>()
    private val syncedActiveShareIds = MutableStateFlow<Set<BeaconId>>(emptySet())
    private val localSharingRoomIds = MutableStateFlow<Set<RoomId>>(emptySet())
    override val sharingRoomIds: StateFlow<Set<RoomId>> = localSharingRoomIds
@@ -139,9 +141,9 @@ class DefaultActiveLiveLocationShareManager(
    }

    override suspend fun onLocationUpdate(location: Location) {
-        val activeSharesCount = localSharingRoomIds.value.size
-        Timber.d("ActiveLiveLocationShareManager received location update for $activeSharesCount active share(s)")
-        localSharingRoomIds.value.forEach { roomId ->
+        val active = stopExpiredShares()
+        Timber.d("ActiveLiveLocationShareManager received location update for ${active.size} active share(s)")
+        active.forEach { roomId ->
            Timber.d("ActiveLiveLocationShareManager sending location to room $roomId")
            sendLiveLocation(roomId, location)
                .onFailure {
@@ -150,6 +152,19 @@ class DefaultActiveLiveLocationShareManager(
        }
    }

+    private suspend fun stopExpiredShares(): List<RoomId> {
+        val nowMillis = clock.epochMillis()
+        val (expired, active) = localSharingRoomIds.value.partition { roomId ->
+            val timeout = timeouts[roomId] ?: return@partition false
+            timeout.expiresAt.toEpochMilliseconds() <= nowMillis
+        }
+        expired.forEach { roomId ->
+            Timber.d("ActiveLiveLocationShareManager location tick detected expired share for room $roomId, stopping")
+            stopShare(roomId)
+        }
+        return active
+    }
+
    private suspend fun sendLiveLocation(roomId: RoomId, location: Location): Result<Unit> {
        val room = cachedRooms.getOrPut(roomId) {
            matrixClient.getJoinedRoom(roomId) ?: return Result.failure(IllegalStateException("No room found for $roomId"))
@@ -192,20 +207,21 @@ class DefaultActiveLiveLocationShareManager(
    }

    private fun scheduleTimeout(roomId: RoomId, expiresAt: Instant) {
-        timeoutJobs.remove(roomId)?.cancel()
+        timeouts.remove(roomId)?.job?.cancel()
        val delayMillis = expiresAt.toEpochMilliseconds() - clock.epochMillis()
-        timeoutJobs[roomId] = matrixClient.sessionCoroutineScope.launch {
+        val job = matrixClient.sessionCoroutineScope.launch {
            delay(delayMillis)
            stopShare(roomId)
                .onFailure { error ->
                    Timber.e(error, "ActiveLiveLocationShareManager failed to stop timed out share for room $roomId")
                }
        }
+        timeouts[roomId] = Timeout(expiresAt = expiresAt, job = job)
    }

    private suspend fun stopLocalShare(roomId: RoomId) {
        Timber.d("ActiveLiveLocationShareManager stop local share in $roomId")
-        timeoutJobs.remove(roomId)?.cancel()
+        timeouts.remove(roomId)?.job?.cancel()
        val wasSharing = localSharingRoomIds.getAndUpdate { it - roomId }.isNotEmpty()
        cachedRooms.remove(roomId)?.close()
        liveLocationStore.removeLiveLocationExpiry(roomId)
@@ -220,11 +236,11 @@ class DefaultActiveLiveLocationShareManager(
        sessionObserver.removeListener(sessionListener)
        coordinator.unregister(matrixClient.sessionId)
        liveLocationStore.clear()
+        timeouts.values.forEach { it.job.cancel() }
+        timeouts.clear()
        for (room in cachedRooms.values) {
            room.close()
-            timeoutJobs[room.roomId]?.cancel()
        }
-        timeoutJobs.clear()
        cachedRooms.clear()
        localSharingRoomIds.value = emptySet()
        syncedActiveShareIds.value = emptySet()
@@ -373,6 +373,53 @@ class DefaultActiveLiveLocationShareManagerTest {
        }
    }

+    @Test
+    fun `location update after expiry stops the share and does not send location`() = runTest {
+        val liveLocationStore = createInMemoryLiveLocationStore()
+        val beaconInfoUpdates = MutableSharedFlow<BeaconInfoUpdate>(replay = 1)
+        val stopLiveLocationShareResult = lambdaRecorder<Result<Unit>> { Result.success(Unit) }
+        val sendLiveLocationResult = lambdaRecorder<String, Result<Unit>> { _ -> Result.success(Unit) }
+        val clock = FakeSystemClock(epochMillisResult = 1_000L)
+        val manager = createManager(
+            client = FakeMatrixClient(
+                sessionId = A_SESSION_ID,
+                sessionCoroutineScope = backgroundScope,
+                ownBeaconInfoUpdates = beaconInfoUpdates,
+            ).apply {
+                givenGetRoomResult(
+                    A_ROOM_ID,
+                    FakeJoinedRoom(
+                        startLiveLocationShareResult = { Result.success(AN_EVENT_ID) },
+                        stopLiveLocationShareResult = stopLiveLocationShareResult,
+                        sendLiveLocationResult = sendLiveLocationResult,
+                    ),
+                )
+            },
+            coordinator = createCoordinator(),
+            liveLocationStore = liveLocationStore,
+            clock = clock,
+        )
+        advanceUntilIdle()
+
+        val startResult = async { manager.startShare(A_ROOM_ID, 1.minutes) }
+        beaconInfoUpdates.emit(BeaconInfoUpdate(roomId = A_ROOM_ID, beaconId = AN_EVENT_ID, isLive = true))
+        assertThat(startResult.await().isSuccess).isTrue()
+
+        // Advance the clock past the expiry so the next location tick is beyond the deadline.
+        clock.epochMillisResult = 1_000L + 1.minutes.inWholeMilliseconds + 1_000L
+
+        manager.sharingRoomIds.test {
+            assertThat(awaitItem()).containsExactly(A_ROOM_ID)
+            manager.onLocationUpdate(io.element.android.features.location.api.Location(lat = 0.0, lon = 0.0, accuracy = null))
+            assertThat(awaitItem()).isEmpty()
+            advanceUntilIdle()
+        }
+        assertThat(liveLocationStore.getLiveLocationExpiries()).doesNotContainKey(A_ROOM_ID)
+        assert(sendLiveLocationResult).isNeverCalled()
+        // stopLiveLocationShare is called twice: once defensively in startShare, once from the tick-triggered stopShare.
+        assert(stopLiveLocationShareResult).isCalledExactly(2)
+    }
+
    @Test
    fun `session deleted clears local state`() = runTest {
        val startServiceRecorder = lambdaRecorder<Unit> { }