fix(sync2): drop live messages beyond the queue limit and keep this device's within it

.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

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

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<String>,
+    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
+    }

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<String>,
+    index: Int,
+) = queueWindow(queue, index) ?: (emptyList<String>() to 0)

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)))
+        }
 }