fix(sync): drop live messages over the size limit or naming a previous object

.github/nightly-notes/2026-11-07-fix-sync2-live-limits.md

@@ -0,0 +1,2 @@
+## Fixed
+- Linked devices: live messages that break the size or format limits are now dropped on the local network too

app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt

@@ -8,7 +8,9 @@ import io.github.aedev.flow.sync2.crypto.purposeKey
 import io.github.aedev.flow.sync2.engine.Session
 import io.github.aedev.flow.sync2.roster.Role
 import io.github.aedev.flow.sync2.store.Header
+import io.github.aedev.flow.sync2.store.MAX_LIVE_MESSAGE
 import io.github.aedev.flow.sync2.store.ObjectType
+import io.github.aedev.flow.sync2.store.ZERO_HASH
 import io.github.aedev.flow.sync2.store.parseObject
 import io.github.aedev.flow.sync2.store.sealObject
 import kotlinx.coroutines.CancellationException
@@ -26,19 +28,20 @@ internal fun Session.sealLive(
     seq: Long,
 ): ByteArray {
     val epoch = state.epoch
-    val header = Header(ObjectType.LIVE, group, epoch, device.did, seq, ByteArray(32))
+    val header = Header(ObjectType.LIVE, group, epoch, device.did, seq, ZERO_HASH)
     val key = purposeKey(currentKey(), group, epoch, Purpose.LIVE)
     return sealObject(header, key, encodePayload(LiveMessage.encode(message)), device.sigSeed)
 }
 
 /** 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 {
+        if (bytes.size > MAX_LIVE_MESSAGE) return null
         val parsed = parseObject(bytes, group)
         val header = parsed.header
         val sender = SyncBytes.toHex(header.sender)
         val member = state.members[sender]?.takeIf { it.role == Role.DEVICE }
-        if (header.type != ObjectType.LIVE || sender == me || member == null) {
+        if (header.type != ObjectType.LIVE || !header.prev.contentEquals(ZERO_HASH) || sender == me || member == null) {
             null
         } else {
             parsed.verify(member.sig)

app/src/test/java/io/github/aedev/flow/sync2/LiveLimitsTest.kt

@@ -0,0 +1,59 @@
+package io.github.aedev.flow.sync2
+
+import android.app.Application
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import io.github.aedev.flow.sync.crypto.SyncBytes
+import io.github.aedev.flow.sync.crypto.SyncCrypto
+import io.github.aedev.flow.sync2.crypto.Purpose
+import io.github.aedev.flow.sync2.crypto.encodePayload
+import io.github.aedev.flow.sync2.crypto.purposeKey
+import io.github.aedev.flow.sync2.engine.Session
+import io.github.aedev.flow.sync2.live.LiveMessage
+import io.github.aedev.flow.sync2.live.openLive
+import io.github.aedev.flow.sync2.live.sealLive
+import io.github.aedev.flow.sync2.store.Header
+import io.github.aedev.flow.sync2.store.MAX_LIVE_MESSAGE
+import io.github.aedev.flow.sync2.store.ObjectType
+import io.github.aedev.flow.sync2.store.sealObject
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.annotation.Config
+
+/** A live message is at most 64 KiB and has no `prev` (spec/store.md), on every transport. */
+@RunWith(AndroidJUnit4::class)
+@Config(sdk = [34], application = Application::class)
+class LiveLimitsTest {
+    @Test
+    fun a_live_message_over_the_limit_is_dropped() =
+        runTest {
+            val g = LinkedGroup(this).apply { create() }
+            val fromA = Session.open(g.deviceDb(), g.store, g.a)
+            val toB = Session.open(g.deviceDb(), g.store, g.b)
+            assertNotNull(toB.openLive(fromA.sealLive(LiveMessage.Now("v", title = "t"), 1)))
+
+            val title = SyncBytes.toHex(SyncCrypto.randomBytes(100 * 1024))
+            val big = fromA.sealLive(LiveMessage.Now("v", title = title), 2)
+
+            assertTrue(big.size > MAX_LIVE_MESSAGE)
+            assertNull(toB.openLive(big))
+        }
+
+    @Test
+    fun a_live_message_naming_a_previous_object_is_dropped() =
+        runTest {
+            val g = LinkedGroup(this).apply { create() }
+            val toB = Session.open(g.deviceDb(), g.store, g.b)
+            val key = purposeKey(g.epochKey, g.group, 0, Purpose.LIVE)
+            val content = encodePayload(LiveMessage.encode(LiveMessage.Now("v")))
+
+            fun envelope(prev: ByteArray) = sealObject(Header(ObjectType.LIVE, g.group, 0, g.a.did, 1, prev), key, content, g.a.sigSeed)
+
+            assertEquals(1L, toB.openLive(envelope(ByteArray(32)))?.seq)
+            assertNull(toB.openLive(envelope(ByteArray(32) { 1 })))
+        }
+}