Add a header to all serialised data & switch to compatibility serializer. (#294)

Add a header to all serialised data & switch to compatibility serializer

core/src/main/kotlin/net/corda/core/serialization/DefaultKryoCustomizer.kt

@@ -1,6 +1,8 @@
 package net.corda.core.serialization
 
 import com.esotericsoftware.kryo.Kryo
+import com.esotericsoftware.kryo.serializers.CompatibleFieldSerializer
+import com.esotericsoftware.kryo.serializers.FieldSerializer
 import com.esotericsoftware.kryo.util.MapReferenceResolver
 import de.javakaffee.kryoserializers.ArraysAsListSerializer
 import de.javakaffee.kryoserializers.UnmodifiableCollectionsSerializer
@@ -25,9 +27,14 @@ object DefaultKryoCustomizer {
         ServiceLoader.load(CordaPluginRegistry::class.java).toList().filter { it.customizeSerialization(customization) }
     }
 
-    // TODO: move all register() to addDefaultSerializer()
     fun customize(kryo: Kryo): Kryo {
         return kryo.apply {
+            // Store a little schema of field names in the stream the first time a class is used which increases tolerance
+            // for change to a class.
+            setDefaultSerializer(CompatibleFieldSerializer::class.java)
+            // Take the safest route here and allow subclasses to have fields named the same as super classes.
+            fieldSerializerConfig.setCachedFieldNameStrategy(FieldSerializer.CachedFieldNameStrategy.EXTENDED)
+
             // Allow construction of objects using a JVM backdoor that skips invoking the constructors, if there is no
             // no-arg constructor available.
             instantiatorStrategy = Kryo.DefaultInstantiatorStrategy(StdInstantiatorStrategy())

core/src/main/kotlin/net/corda/core/serialization/Kryo.kt

@@ -76,10 +76,19 @@ class SerializedBytes<T : Any>(bytes: ByteArray, val internalOnly: Boolean = fal
     fun writeToFile(path: Path): Path = Files.write(path, bytes)
 }
 
+// "corda" + majorVersionByte + minorVersionMSB + minorVersionLSB
+private val KryoHeaderV0_1: OpaqueBytes = OpaqueBytes("corda\u0000\u0000\u0001".toByteArray())
+
 // Some extension functions that make deserialisation convenient and provide auto-casting of the result.
 fun <T : Any> ByteArray.deserialize(kryo: Kryo = threadLocalP2PKryo()): T {
-    @Suppress("UNCHECKED_CAST")
-    return kryo.readClassAndObject(Input(this)) as T
+    Input(this).use {
+        val header = OpaqueBytes(it.readBytes(8))
+        if (header != KryoHeaderV0_1) {
+            throw KryoException("Serialized bytes header does not match any known format.")
+        }
+        @Suppress("UNCHECKED_CAST")
+        return kryo.readClassAndObject(it) as T
+    }
 }
 
 fun <T : Any> OpaqueBytes.deserialize(kryo: Kryo = threadLocalP2PKryo()): T {
@@ -114,6 +123,7 @@ object SerializedBytesSerializer : Serializer<SerializedBytes<Any>>() {
 fun <T : Any> T.serialize(kryo: Kryo = threadLocalP2PKryo(), internalOnly: Boolean = false): SerializedBytes<T> {
     val stream = ByteArrayOutputStream()
     Output(stream).use {
+        it.writeBytes(KryoHeaderV0_1.bytes)
         kryo.writeClassAndObject(it, this)
     }
     return SerializedBytes(stream.toByteArray(), internalOnly)

core/src/test/kotlin/net/corda/core/flows/BroadcastTransactionFlowTest.kt

@@ -1,6 +1,5 @@
 package net.corda.core.flows
 
-import com.esotericsoftware.kryo.io.Input
 import com.pholser.junit.quickcheck.From
 import com.pholser.junit.quickcheck.Property
 import com.pholser.junit.quickcheck.generator.GenerationStatus
@@ -8,7 +7,7 @@ import com.pholser.junit.quickcheck.generator.Generator
 import com.pholser.junit.quickcheck.random.SourceOfRandomness
 import com.pholser.junit.quickcheck.runner.JUnitQuickcheck
 import net.corda.contracts.testing.SignedTransactionGenerator
-import net.corda.core.serialization.createKryo
+import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
 import net.corda.flows.BroadcastTransactionFlow.NotifyTxRequest
 import org.junit.runner.RunWith
@@ -25,9 +24,8 @@ class BroadcastTransactionFlowTest {
 
     @Property
     fun serialiseDeserialiseOfNotifyMessageWorks(@From(NotifyTxRequestMessageGenerator::class) message: NotifyTxRequest) {
-        val kryo = createKryo()
         val serialized = message.serialize().bytes
-        val deserialized = kryo.readClassAndObject(Input(serialized))
+        val deserialized = serialized.deserialize<NotifyTxRequest>()
         assertEquals(deserialized, message)
     }
 }

node-schemas/src/main/kotlin/net/corda/node/services/vault/schemas/VaultSchema.kt

@@ -3,7 +3,6 @@ package net.corda.node.services.vault.schemas
 import io.requery.*
 import net.corda.core.node.services.Vault
 import net.corda.core.schemas.requery.Requery
-import net.corda.core.schemas.requery.converters.InstantConverter
 import java.time.Instant
 
 object VaultSchema {
@@ -50,7 +49,7 @@ object VaultSchema {
 
         /** refers to serialized transaction Contract State */
         // TODO: define contract state size maximum size and adjust length accordingly
-        @get:Column(name = "contract_state", length = 10000)
+        @get:Column(name = "contract_state", length = 100000)
         var contractState: ByteArray
 
         /** state lifecycle: unconsumed, consumed */

samples/irs-demo/src/main/resources/simulation/trade.json

@@ -1,6 +1,6 @@
 {
   "fixedLeg": {
-    "fixedRatePayer": "2eFzn8gRQq7nNgypMCjKik4w8i565TM3xBmp85eefhG1c24VSj5",
+    "fixedRatePayer": "3ThWzJauCq7qLrcX4KuKHxKnxZ6HoxnxU7pFL1HwfCkCJLUfTJ9zN92oxRLxnw",
     "notional": {
       "quantity": 2500000000,
       "token": "USD"
@@ -25,7 +25,7 @@
     "interestPeriodAdjustment": "Adjusted"
   },
   "floatingLeg": {
-    "floatingRatePayer": "2eFzn8gJj7xcdBxExC7XEiiX36dw6HfG3MCpjMt2CaejwUnfAxb",
+    "floatingRatePayer": "3ThWzJauCq7qLrcX4KndaoA3TXWtoLzUSmQ9ia8xVWxHTmVGjXXXFXm9R9Wh2T",
     "notional": {
       "quantity": 2500000000,
       "token": "USD"