Resolve a problem when Exceptions with suppression passed through Kryo serialization (#1280)

Also added some unit tests that expose the problem

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/CordaClassResolver.kt

@@ -77,6 +77,7 @@ class CordaClassResolver(serializationContext: SerializationContext) : DefaultCl
                 objectInstance != null -> KotlinObjectSerializer(objectInstance)
                 kotlin.jvm.internal.Lambda::class.java.isAssignableFrom(type) -> // Kotlin lambdas extend this class and any captured variables are stored in synthetic fields
                     FieldSerializer<Any>(kryo, type).apply { setIgnoreSyntheticFields(false) }
+                Throwable::class.java.isAssignableFrom(type) -> ThrowableSerializer(kryo, type)
                 else -> kryo.getDefaultSerializer(type)
             }
             return register(Registration(type, serializer, NAME.toInt()))

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/Kryo.kt

@@ -1,8 +1,11 @@
 package net.corda.nodeapi.internal.serialization
 
 import com.esotericsoftware.kryo.*
+import com.esotericsoftware.kryo.factories.ReflectionSerializerFactory
 import com.esotericsoftware.kryo.io.Input
 import com.esotericsoftware.kryo.io.Output
+import com.esotericsoftware.kryo.serializers.CompatibleFieldSerializer
+import com.esotericsoftware.kryo.serializers.FieldSerializer
 import com.esotericsoftware.kryo.util.MapReferenceResolver
 import net.corda.core.contracts.*
 import net.corda.core.crypto.Crypto
@@ -565,4 +568,46 @@ object X509CertificateSerializer : Serializer<X509CertificateHolder>() {
     }
 }
 
-fun Kryo.serializationContext(): SerializeAsTokenContext? = context.get(serializationContextKey) as? SerializeAsTokenContext
+fun Kryo.serializationContext(): SerializeAsTokenContext? = context.get(serializationContextKey) as? SerializeAsTokenContext
+
+/**
+ * For serializing instances if [Throwable] honoring the fact that [java.lang.Throwable.suppressedExceptions]
+ * might be un-initialized/empty.
+ * In the absence of this class [CompatibleFieldSerializer] will be used which will assign a *new* instance of
+ * unmodifiable collection to [java.lang.Throwable.suppressedExceptions] which will fail some sentinel identity checks
+ * e.g. in [java.lang.Throwable.addSuppressed]
+ */
+@ThreadSafe
+class ThrowableSerializer<T>(kryo: Kryo, type: Class<T>) : Serializer<Throwable>(false, true) {
+
+    private companion object {
+        private val suppressedField = Throwable::class.java.getDeclaredField("suppressedExceptions")
+
+        private val sentinelValue = let {
+            val sentinelField = Throwable::class.java.getDeclaredField("SUPPRESSED_SENTINEL")
+            sentinelField.isAccessible = true
+            sentinelField.get(null)
+        }
+
+        init {
+            suppressedField.isAccessible = true
+        }
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private val delegate: Serializer<Throwable> = ReflectionSerializerFactory.makeSerializer(kryo, FieldSerializer::class.java, type) as Serializer<Throwable>
+
+    override fun write(kryo: Kryo, output: Output, throwable: Throwable) {
+        delegate.write(kryo, output, throwable)
+    }
+
+    override fun read(kryo: Kryo, input: Input, type: Class<Throwable>): Throwable {
+        val throwableRead = delegate.read(kryo, input, type)
+        if(throwableRead.suppressed.isEmpty()) {
+            throwableRead.setSuppressedToSentinel()
+        }
+        return throwableRead
+    }
+
+    private fun Throwable.setSuppressedToSentinel() = suppressedField.set(this, sentinelValue)
+}

node-api/src/test/kotlin/net/corda/nodeapi/internal/serialization/KryoTests.kt

@@ -7,6 +7,7 @@ import com.esotericsoftware.kryo.io.Output
 import com.google.common.primitives.Ints
 import net.corda.core.contracts.PrivacySalt
 import net.corda.core.crypto.*
+import net.corda.core.internal.FetchDataFlow
 import net.corda.core.serialization.*
 import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.sequence
@@ -25,6 +26,7 @@ import java.io.ByteArrayInputStream
 import java.io.InputStream
 import java.time.Instant
 import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 
 class KryoTests : TestDependencyInjectionBase() {
@@ -235,4 +237,42 @@ class KryoTests : TestDependencyInjectionBase() {
                 SerializationContext.UseCase.P2P)
         pt.serialize(factory, context)
     }
-}
+
+    @Test
+    fun `serialize - deserialize Exception with suppressed`() {
+        val exception = IllegalArgumentException("fooBar")
+        val toBeSuppressedOnSenderSide = IllegalStateException("bazz1")
+        exception.addSuppressed(toBeSuppressedOnSenderSide)
+        val exception2 = exception.serialize(factory, context).deserialize(factory, context)
+        assertEquals(exception.message, exception2.message)
+
+        assertEquals(1, exception2.suppressed.size)
+        assertNotNull({ exception2.suppressed.find { it.message == toBeSuppressedOnSenderSide.message  }})
+
+        val toBeSuppressedOnReceiverSide = IllegalStateException("bazz2")
+        exception2.addSuppressed(toBeSuppressedOnReceiverSide)
+        assertTrue { exception2.suppressed.contains(toBeSuppressedOnReceiverSide) }
+        assertEquals(2, exception2.suppressed.size)
+    }
+
+    @Test
+    fun `serialize - deserialize Exception no suppressed`() {
+        val exception = IllegalArgumentException("fooBar")
+        val exception2 = exception.serialize(factory, context).deserialize(factory, context)
+        assertEquals(exception.message, exception2.message)
+        assertEquals(0, exception2.suppressed.size)
+
+        val toBeSuppressedOnReceiverSide = IllegalStateException("bazz2")
+        exception2.addSuppressed(toBeSuppressedOnReceiverSide)
+        assertEquals(1, exception2.suppressed.size)
+        assertTrue { exception2.suppressed.contains(toBeSuppressedOnReceiverSide) }
+    }
+
+    @Test
+    fun `serialize - deserialize HashNotFound`() {
+        val randomHash = SecureHash.randomSHA256()
+        val exception = FetchDataFlow.HashNotFound(randomHash)
+        val exception2 = exception.serialize(factory, context).deserialize(factory, context)
+        assertEquals(randomHash, exception2.requested)
+    }
+}