From 1efb20468b8fe53f9e8b43450d84fad6a565edef Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:31:46 +0200 Subject: [PATCH] fix(sync2): drop live messages beyond the queue limit and keep this device's within it --- .../2026-11-02-fix-sync2-live-queue.md | 2 ++ .../aedev/flow/sync2/live/LiveEnvelopes.kt | 4 ++-- .../aedev/flow/sync2/live/LiveMessage.kt | 13 ++++++++++++ .../aedev/flow/sync2/live/NowPlaying.kt | 17 ++++++++++----- .../aedev/flow/sync2/LiveChannelTest.kt | 21 +++++++++++++++++++ 5 files changed, 50 insertions(+), 7 deletions(-) create mode 100644 .github/nightly-notes/2026-11-02-fix-sync2-live-queue.md diff --git a/.github/nightly-notes/2026-11-02-fix-sync2-live-queue.md b/.github/nightly-notes/2026-11-02-fix-sync2-live-queue.md new file mode 100644 index 000000000..df31a6379 --- /dev/null +++ b/.github/nightly-notes/2026-11-02-fix-sync2-live-queue.md @@ -0,0 +1,2 @@ +## Fixed +- Linked devices: Flow Desktop now shows what plays here past the 50th item of a queue diff --git a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt index 737e73633..503a00171 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt @@ -31,7 +31,7 @@ internal fun Session.sealLive( return sealObject(header, key, encodePayload(LiveMessage.encode(message)), device.sigSeed) } -/** Opens a live envelope signed by another current device of this group, or null for anything else. */ +/** Opens a valid live message signed by another current device of this group, or null for anything else. */ internal suspend fun Session.openLive(bytes: ByteArray): LiveReceived? = try { val parsed = parseObject(bytes, group) @@ -43,7 +43,7 @@ internal suspend fun Session.openLive(bytes: ByteArray): LiveReceived? = } else { parsed.verify(member.sig) val content = decodePayload(parsed.open(purposeKey(epochKey(header.epoch), group, header.epoch, Purpose.LIVE))) - LiveReceived(sender, header.seq, LiveMessage.decode(content)) + LiveMessage.decode(content).takeIf { it.isValid() }?.let { LiveReceived(sender, header.seq, it) } } } catch (e: CancellationException) { throw e diff --git a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveMessage.kt b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveMessage.kt index 8c1bb21ad..bb3398ae9 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveMessage.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveMessage.kt @@ -92,3 +92,16 @@ sealed interface LiveMessage { fun decode(content: ByteArray): LiveMessage = json.decodeFromString(serializer(), content.decodeToString()) } } + +private fun fits( + queue: List, + index: Int, +) = queue.size <= LiveMessage.QUEUE_LIMIT && (queue.isEmpty() || index in queue.indices) + +/** Whether the fields are within the limits spec/live.md sets. */ +internal fun LiveMessage.isValid(): Boolean = + when (this) { + is LiveMessage.Now -> vol in 0..100 && fits(queue, index) + is LiveMessage.Play -> fits(queue, index) + else -> true + } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/live/NowPlaying.kt b/app/src/main/java/io/github/aedev/flow/sync2/live/NowPlaying.kt index 70d5b8b87..61a2f908c 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/live/NowPlaying.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/live/NowPlaying.kt @@ -111,6 +111,7 @@ class NowPlaying val vol = volume() if (track != null && musicInUse()) { val queue = EnhancedMusicPlayerManager.queue.value + val (ids, index) = window(queue.map { it.videoId }, queue.indexOfFirst { it.videoId == track.videoId }) return LiveMessage.Now( video = track.videoId, title = track.title, @@ -120,12 +121,12 @@ class NowPlaying dur = music.duration, state = if (music.isPlaying) LiveMessage.PLAYING else LiveMessage.PAUSED, vol = vol, - queue = queue.take(LiveMessage.QUEUE_LIMIT).map { it.videoId }, - index = queue.indexOfFirst { it.videoId == track.videoId }.coerceAtLeast(0), + queue = ids, + index = index, ) } val current = GlobalPlayerState.currentVideo.value ?: return LiveMessage.Now(video = null) - val queue = video.queueVideos.value + val (ids, index) = window(video.queueVideos.value.map { it.id }, video.currentQueueIndexState.value) return LiveMessage.Now( video = current.id, title = current.title, @@ -134,8 +135,8 @@ class NowPlaying dur = video.getDuration().coerceAtLeast(0), state = if (GlobalPlayerState.playerState.value.isPlaying) LiveMessage.PLAYING else LiveMessage.PAUSED, vol = vol, - queue = queue.take(LiveMessage.QUEUE_LIMIT).map { it.id }, - index = video.currentQueueIndexState.value.coerceAtLeast(0), + queue = ids, + index = index, ) } @@ -146,3 +147,9 @@ class NowPlaying return audio.getStreamVolume(AudioManager.STREAM_MUSIC) * 100 / max } } + +/** The part of [queue] a `now` names, around [index], or none when [index] is outside it. */ +private fun window( + queue: List, + index: Int, +) = queueWindow(queue, index) ?: (emptyList() to 0) diff --git a/app/src/test/java/io/github/aedev/flow/sync2/LiveChannelTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/LiveChannelTest.kt index ba9704504..5d8a09c8b 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/LiveChannelTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/LiveChannelTest.kt @@ -8,10 +8,13 @@ import io.github.aedev.flow.sync.crypto.SyncBytes import io.github.aedev.flow.sync2.crypto.MemberSecrets import io.github.aedev.flow.sync2.data.LinkedDatabase import io.github.aedev.flow.sync2.device.DeviceKeys +import io.github.aedev.flow.sync2.engine.Session import io.github.aedev.flow.sync2.live.LiveChannel import io.github.aedev.flow.sync2.live.LiveMessage import io.github.aedev.flow.sync2.live.LiveReceived import io.github.aedev.flow.sync2.live.LiveTransport +import io.github.aedev.flow.sync2.live.openLive +import io.github.aedev.flow.sync2.live.sealLive import io.mockk.coEvery import io.mockk.mockk import kotlinx.coroutines.CoroutineStart @@ -20,6 +23,7 @@ import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.toList import kotlinx.coroutines.test.runTest import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull import org.junit.Assert.assertTrue import org.junit.Test import org.junit.runner.RunWith @@ -104,4 +108,21 @@ class LiveChannelTest { assertEquals("only B's presence arrives", SyncBytes.toHex(g.b.did), heard.await().single().sender) } + + @Test + fun a_message_beyond_the_limits_of_the_spec_is_dropped() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (a, b) = Session.open(g.deviceDb(), g.store, g.a) to Session.open(g.deviceDb(), g.store, g.b) + + fun play( + size: Int, + index: Int, + ) = LiveMessage.Play(SyncBytes.toHex(g.a.did), "v0", 0, false, null, List(size) { "v$it" }, index, LiveMessage.PLAYING) + + assertEquals(play(50, 49), a.openLive(b.sealLive(play(50, 49), 1))?.message) + assertNull("51 videos", a.openLive(b.sealLive(play(51, 0), 2))) + assertNull("an index outside the queue", a.openLive(b.sealLive(play(3, 3), 3))) + assertNull("a volume above 100", a.openLive(b.sealLive(now.copy(vol = 101), 4))) + } }