Fix stale SyncSetDeltaPacket rejection by tracking actual change IDs

surf-cloud-core/surf-cloud-core-client/src/main/kotlin/dev/slne/surf/cloud/core/client/sync/SyncRegistryImpl.kt

@@ -62,12 +62,12 @@ class SyncRegistryImpl : CommonSyncRegistryImpl() {
         lastChangeIds[packet.setId] = packet.changeId
     }
 
-    fun applyBatchSyncSets(bulk: List<Pair<String, Set<Any?>>>) {
-        bulk.forEach { (id, snapshot) ->
+    fun applyBatchSyncSets(bulk: List<Triple<String, Set<Any?>, Long>>) {
+        bulk.forEach { (id, snapshot, changeId) ->
             val set = getSet<Any?>(id)
             if (set != null) {
                 set.addAllInternal(snapshot)
-                lastChangeIds[id] = Long.MAX_VALUE // Reset change ID to max after bulk update
+                lastChangeIds[id] = changeId
             }
         }
     }

surf-cloud-core/surf-cloud-core-common/src/main/kotlin/dev/slne/surf/cloud/core/common/netty/network/protocol/synchronizing/ClientboundBatchSyncSetPacket.kt

@@ -20,17 +20,17 @@
     ConnectionProtocol.SYNCHRONIZING,
     handlerMode = PacketHandlerMode.DEFAULT
 )
 class ClientboundBatchSyncSetPacket : NettyPacket, InternalNettyPacket<ClientSynchronizingPacketListener> {
     companion object {
         private val log = logger()
         val STREAM_CODEC =
             packetCodec(ClientboundBatchSyncSetPacket::write, ::ClientboundBatchSyncSetPacket)
     }
 
-    val syncSets: List<Pair<String, Set<Any?>>>
+    val syncSets: List<Triple<String, Set<Any?>, Long>>
 
     constructor(syncValues: Map<String, SyncSetImpl<*>>) {
-        this.syncSets = syncValues.map { (key, value) -> key to value.toSet() }
+        this.syncSets = syncValues.map { (key, value) -> Triple(key, value.toSet(), value.currentChangeId) }
     }
 
     private constructor(buf: SurfByteBuf) {
@@ -43,10 +43,13 @@
             val syncSet = CommonSyncRegistryImpl.instance.getSet<Any?>(syncId)
             if (syncSet == null) {
                 buf.skipBytes(syncSize)
+                buf.readLong() // Skip the change ID as well
                 unknownSyncValues.add(syncId)
                 null
             } else {
-                syncId to syncSet.codec.decode(buf)
+                val set = syncSet.codec.decode(buf)
+                val changeId = buf.readLong()
+                Triple(syncId, set, changeId)
             }
         }.filterNotNull()
 
@@ -57,7 +60,7 @@
     }
 
     private fun write(buf: SurfByteBuf) {
-        buf.writeCollection(syncSets) { buf, (syncId, set) ->
+        buf.writeCollection(syncSets) { buf, (syncId, set, changeId) ->
             buf.writeUtf(syncId)
 
             // Reserve 4 bytes for length
@@ -72,6 +75,9 @@
 
             // Write the actual length of the encoded value
             buf.setInt(lengthIndex, endIndex - startIndex)
+
+            // Write the change ID after the set data
+            buf.writeLong(changeId)
         }
     }
 

surf-cloud-core/surf-cloud-core-common/src/main/kotlin/dev/slne/surf/cloud/core/common/sync/SyncSetImpl.kt

@@ -29,6 +29,8 @@ class SyncSetImpl<T>(override val id: String, val valueCodec: StreamCodec<SurfBy
     private val listeners = CopyOnWriteArrayList<SyncSetListener<T>>()
     private val changeCounter = AtomicLong()
 
+    val currentChangeId: Long
+        get() = changeCounter.get()
 
     init {
         CommonSyncRegistryImpl.instance.register(this)