Fix: GroupUpdated send jobs could not be read back after a restart

app/src/main/java/org/session/libsession/messaging/jobs/AttachmentUploadJob.kt

@@ -1,6 +1,5 @@
 package org.session.libsession.messaging.jobs
 
-import com.esotericsoftware.kryo.Kryo
 import com.esotericsoftware.kryo.io.Input
 import com.esotericsoftware.kryo.io.Output
 import dagger.assisted.Assisted
@@ -248,8 +247,7 @@ class AttachmentUploadJob @AssistedInject constructor(
     }
 
     override fun serialize(): Data {
-        val kryo = Kryo()
-        kryo.isRegistrationRequired = false
+        val kryo = jobKryo()
         val serializedMessage = ByteArray(4096)
         val output = Output(serializedMessage, Job.MAX_BUFFER_SIZE_BYTES)
         kryo.writeClassAndObject(output, message)
@@ -277,8 +275,7 @@ class AttachmentUploadJob @AssistedInject constructor(
 
         override fun create(data: Data): AttachmentUploadJob? {
             val serializedMessage = data.getByteArray(MESSAGE_KEY)
-            val kryo = Kryo()
-            kryo.isRegistrationRequired = false
+            val kryo = jobKryo()
             val input = Input(serializedMessage)
             val message: Message
             try {

app/src/main/java/org/session/libsession/messaging/jobs/JobKryo.kt

@@ -0,0 +1,39 @@
+package org.session.libsession.messaging.jobs
+
+import com.esotericsoftware.kryo.Kryo
+import com.esotericsoftware.kryo.Serializer
+import com.esotericsoftware.kryo.io.Input
+import com.esotericsoftware.kryo.io.Output
+import com.google.protobuf.MessageLite
+
+/**
+ * The [Kryo] instance used to persist jobs, and to read them back after a restart.
+ *
+ * Every job that persists a message must use this rather than a bare [Kryo]: the two directions
+ * have to agree on the serializers, and a job that writes with one configuration and reads with
+ * another is silently dropped on restart rather than failing at the point of the mistake.
+ */
+internal fun jobKryo(): Kryo = Kryo().apply {
+    isRegistrationRequired = false
+    addDefaultSerializer(MessageLite::class.java, ProtobufSerializer())
+}
+
+/**
+ * Kryo builds objects field by field, and parts of a protobuf message's object graph have no
+ * constructor it can call — so it writes one happily and then throws ("Class cannot be created") on
+ * the way back in, taking the job that held it with it. Persist the wire format the message can
+ * rebuild itself from instead.
+ */
+private class ProtobufSerializer : Serializer<MessageLite>() {
+    override fun write(kryo: Kryo, output: Output, message: MessageLite) {
+        val bytes = message.toByteArray()
+        output.writeVarInt(bytes.size, true)
+        output.writeBytes(bytes)
+    }
+
+    override fun read(kryo: Kryo, input: Input, type: Class<out MessageLite>): MessageLite {
+        val bytes = input.readBytes(input.readVarInt(true))
+
+        return type.getMethod("parseFrom", ByteArray::class.java).invoke(null, bytes) as MessageLite
+    }
+}

app/src/main/java/org/session/libsession/messaging/jobs/MessageSendJob.kt

@@ -1,6 +1,5 @@
 package org.session.libsession.messaging.jobs
 
-import com.esotericsoftware.kryo.Kryo
 import com.esotericsoftware.kryo.io.Input
 import com.esotericsoftware.kryo.io.Output
 import dagger.assisted.Assisted
@@ -157,8 +156,7 @@ class MessageSendJob @AssistedInject constructor(
     }
 
     override fun serialize(): Data {
-        val kryo = Kryo()
-        kryo.isRegistrationRequired = false
+        val kryo = jobKryo()
         // Message
         val messageOutput = Output(ByteArray(4096), MAX_BUFFER_SIZE_BYTES)
         kryo.writeClassAndObject(messageOutput, message)
@@ -192,8 +190,7 @@ class MessageSendJob @AssistedInject constructor(
         override fun create(data: Data): MessageSendJob? {
             val serializedMessage = data.getByteArray(MESSAGE_KEY)
             val serializedDestination = data.getByteArray(DESTINATION_KEY)
-            val kryo = Kryo()
-            kryo.isRegistrationRequired = false
+            val kryo = jobKryo()
             // Message
             val messageInput = Input(serializedMessage)
             val message: Message

app/src/test/java/org/session/libsession/messaging/jobs/JobKryoTest.kt

@@ -0,0 +1,36 @@
+package org.session.libsession.messaging.jobs
+
+import com.esotericsoftware.kryo.io.Input
+import com.esotericsoftware.kryo.io.Output
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.session.libsession.messaging.messages.control.GroupUpdated
+import org.session.protos.SessionProtos
+
+class JobKryoTest {
+    @Test
+    fun `GroupUpdated survives a round trip`() {
+        val message = GroupUpdated(
+            SessionProtos.GroupUpdateMessage.newBuilder()
+                .setMemberLeftMessage(SessionProtos.GroupUpdateMemberLeftMessage.getDefaultInstance())
+                .build()
+        ).apply {
+            sentTimestamp = 1_234_567_890L
+        }
+
+        val restored = roundTrip(message)
+
+        assertTrue(restored is GroupUpdated)
+        assertEquals(message.inner, (restored as GroupUpdated).inner)
+        assertEquals(message.sentTimestamp, restored.sentTimestamp)
+    }
+
+    private fun roundTrip(value: Any): Any {
+        val output = Output(ByteArray(4096), Job.MAX_BUFFER_SIZE_BYTES)
+        jobKryo().writeClassAndObject(output, value)
+        output.close()
+
+        return jobKryo().readClassAndObject(Input(output.toBytes()))
+    }
+}