Fix serialization of subclasses (#92)

* Add integration test for catching Exception instead of StepError

This is the reproduction of an error reported by a user. Because
StepError contains additional fields to a generic Exception,
deserialization fails.

* Add integration test for the general case of deserializing a subclass

This shows that the issue is not limited to StepError/Exception.

* Fix deserialization of subclasses

Previously we were deserializing step data based on the T provided by the compiler.
However, this could be lossy if the step data was actually a subclass of
T with additional fields and is semantically incorrect if we try to
deserialize it as the parent class.

To remedy this, we need to know what the actual class of the step data
was before it was memoized with the Inngest server. We are adding this
as an additional field `class` on the JSON when serializing, then
reading (and deleting) this field and using it with reflection to
deserialize using the proper class.

It might make sense to put the field `class` in some kind of
meta/extension section of the payload to avoid conflict, but it's highly
unlikely a field will legitimately be named class in the user code since
it is a reserved keyword in Java.

This strategy doesn't entirely work for schema evolution because it's
still possible the classes have changed to the point where
deserialization would fail. We should follow up here and probably have a
fallback mechanism of returning a generic hash if deserialization fails.

---------

Co-authored-by: Albert Chae <albertchae@users.noreply.github.com>
Co-authored-by: Riadh <22998716+KiKoS0@users.noreply.github.com>

inngest-spring-boot-demo/src/main/java/com/inngest/springbootdemo/testfunctions/DeserializeSubclassFunction.java

@@ -0,0 +1,49 @@
+package com.inngest.springbootdemo.testfunctions;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.inngest.FunctionContext;
+import com.inngest.InngestFunction;
+import com.inngest.InngestFunctionConfigBuilder;
+import com.inngest.Step;
+import org.jetbrains.annotations.NotNull;
+
+class Dog {
+    @JsonProperty("legs")
+    public int legs;
+
+    public Dog(@JsonProperty("legs") int legs) {
+        this.legs = legs;
+    }
+}
+
+class Corgi extends Dog {
+    @JsonProperty("stumpy")
+    public boolean stumpy;
+
+    public Corgi(@JsonProperty("legs") int legs, @JsonProperty("stumpy") boolean stumpy) {
+        super(legs);
+
+        this.stumpy = stumpy;
+    }
+}
+
+public class DeserializeSubclassFunction extends InngestFunction {
+    @NotNull
+    @Override
+    public InngestFunctionConfigBuilder config(InngestFunctionConfigBuilder builder) {
+        return builder
+            .id("DeserializeSubclassFunction")
+            .name("Deserialize subclass function")
+            .triggerEvent("test/deserialize.subclass")
+            .retries(0);
+    }
+
+    @Override
+    public String execute(FunctionContext ctx, Step step) {
+        Dog corgi = step.run("get-corgi", () -> new Corgi(4, true), Dog.class);
+
+        assert(((Corgi) corgi).stumpy == true);
+
+        return "Successfully cast Corgi";
+    }
+}

inngest-spring-boot-demo/src/main/java/com/inngest/springbootdemo/testfunctions/TryCatchGenericExceptionFunction.java

@@ -0,0 +1,30 @@
+package com.inngest.springbootdemo.testfunctions;
+
+import com.inngest.*;
+import org.jetbrains.annotations.NotNull;
+
+public class TryCatchGenericExceptionFunction extends InngestFunction {
+    @NotNull
+    @Override
+    public InngestFunctionConfigBuilder config(InngestFunctionConfigBuilder builder) {
+        return builder
+            .id("try-catch-deserialize-exception-function")
+            .name("Try Catch Deserialize Exception Function")
+            .triggerEvent("test/try.catch.deserialize.exception")
+            .retries(0);
+    }
+
+    @Override
+    public String execute(FunctionContext ctx, Step step) {
+        try {
+            step.run("fail-step", () -> {
+                throw new CustomException("Something fatally went wrong");
+            }, String.class);
+        } catch (Exception originalException) {
+            Exception e = step.run("handle-error", () -> originalException, Exception.class);
+            return e.getMessage();
+        }
+
+        return "An error should have been thrown and this message should not be returned";
+    }
+}

inngest-spring-boot-demo/src/test/java/com/inngest/springbootdemo/DemoTestConfiguration.java

@@ -22,6 +22,8 @@ protected HashMap<String, InngestFunction> functions() {
         addInngestFunction(functions, new RetriableErrorFunction());
         addInngestFunction(functions, new ZeroRetriesFunction());
         addInngestFunction(functions, new InvokeFailureFunction());
+        addInngestFunction(functions, new DeserializeSubclassFunction());
+        addInngestFunction(functions, new TryCatchGenericExceptionFunction());
         addInngestFunction(functions, new TryCatchRunFunction());
         addInngestFunction(functions, new ThrottledFunction());
         addInngestFunction(functions, new RateLimitedFunction());

inngest-spring-boot-demo/src/test/java/com/inngest/springbootdemo/DeserializationIntegrationTest.java

@@ -0,0 +1,42 @@
+package com.inngest.springbootdemo;
+
+import com.inngest.Inngest;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.LinkedHashMap;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@IntegrationTest
+@Execution(ExecutionMode.CONCURRENT)
+class DeserializationIntegrationTest {
+    @Autowired
+    private DevServerComponent devServer;
+
+    static int sleepTime = 5000;
+
+    @Autowired
+    private Inngest client;
+
+    @Test
+    void testShouldDeserializeSubclassCorrectly() throws Exception {
+        String eventId = InngestFunctionTestHelpers.sendEvent(client, "test/deserialize.subclass").getIds()[0];
+
+        Thread.sleep(sleepTime);
+
+        RunEntry<Object> run = devServer.runsByEvent(eventId).first();
+        Object output = run.getOutput();
+        if (output instanceof LinkedHashMap) {
+            fail("Run threw an exception serialized into a LinkedHashMap:" + output);
+        }
+        String outputString = (String) output;
+
+        assertEquals("Completed", run.getStatus() );
+        assertNotNull(run.getEnded_at());
+
+        assertEquals("Successfully cast Corgi", outputString);
+    }
+}

inngest-spring-boot-demo/src/test/java/com/inngest/springbootdemo/StepErrorsIntegrationTest.java

@@ -6,8 +6,9 @@
 import org.junit.jupiter.api.parallel.ExecutionMode;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
+import java.util.LinkedHashMap;
+
+import static org.junit.jupiter.api.Assertions.*;
 
 @IntegrationTest
 @Execution(ExecutionMode.CONCURRENT)
@@ -50,4 +51,22 @@ void testShouldCatchStepErrorWhenRunThrows() throws Exception {
         assertEquals("Something fatally went wrong", output);
     }
 
+    @Test
+    void testShouldCatchAndDeserializeExceptionWhenRunThrows() throws Exception {
+        String eventId = InngestFunctionTestHelpers.sendEvent(client, "test/try.catch.deserialize.exception").getIds()[0];
+
+        Thread.sleep(sleepTime);
+
+        RunEntry<Object> run = devServer.runsByEvent(eventId).first();
+        Object output = run.getOutput();
+        if (output instanceof LinkedHashMap) {
+            fail("Run threw an exception serialized into a LinkedHashMap:" + output);
+        }
+        String outputString = (String) output;
+
+        assertEquals("Completed", run.getStatus());
+        assertNotNull(run.getEnded_at());
+
+        assertEquals("Something fatally went wrong", outputString);
+    }
 }

inngest/src/main/kotlin/com/inngest/Function.kt

@@ -1,6 +1,9 @@
 package com.inngest
 
 import com.beust.klaxon.Json
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.node.ObjectNode
 import java.util.function.BiFunction
 
 // TODO - Add an abstraction layer between the Function call response and the comm handler response
@@ -229,8 +232,9 @@ internal open class InternalInngestFunction(
             // NOTE - Currently this error could be caught in the user's own function
             // that wraps a
             // step.run() - how can we prevent that or warn?
+
             return StepResult(
-                data = e.data,
+                data = serializeStepData(e.data),
                 id = e.hashedId,
                 name = e.id,
                 op = OpCode.StepRun,
@@ -255,4 +259,23 @@ internal open class InternalInngestFunction(
         // TODO use URL objects for serveUrl instead of strings so we can fetch things like scheme
         return configBuilder.build(client.appId, serveUrl)
     }
+
+    private fun serializeStepData(stepData: Any?): JsonNode? {
+        if (stepData == null) {
+            return stepData
+        }
+
+        val mapper = ObjectMapper()
+        val jsonString = mapper.writeValueAsString(stepData)
+        val readOnlyJson = mapper.readTree(jsonString)
+
+        if (!readOnlyJson.isObject) {
+            // primitives can be serialized directly
+            return readOnlyJson
+        }
+
+        val writeableJson = mapper.readTree(jsonString) as ObjectNode
+        writeableJson.put("class", stepData.javaClass.name)
+        return writeableJson
+    }
 }

inngest/src/main/kotlin/com/inngest/State.kt

@@ -2,6 +2,7 @@ package com.inngest
 
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.node.ObjectNode
 import java.security.MessageDigest
 
 class StateNotFound : Throwable("State not found for id")
@@ -60,8 +61,7 @@ class State(
         val stepResult = node.path("steps").get(hashedId) ?: throw StateNotFound()
 
         if (stepResult.has(fieldName)) {
-            val dataNode = stepResult.get(fieldName)
-            return mapper.treeToValue(dataNode, type)
+            return deserializeStepData(stepResult.get(fieldName), type)
         } else if (stepResult.has("error")) {
             val error = mapper.treeToValue(stepResult.get("error"), StepError::class.java)
             throw error
@@ -71,4 +71,19 @@ class State(
         // TODO - Check the state is actually null
         return null
     }
+
+    private fun <T> deserializeStepData(
+        serializedStepData: JsonNode?,
+        type: Class<T>,
+    ): T? {
+        val mapper = ObjectMapper()
+        if (serializedStepData == null || !serializedStepData.isObject || !serializedStepData.has("class")) {
+            // null and primitives can be deserialized directly
+            return mapper.treeToValue(serializedStepData, type)
+        }
+
+        val writeableJson = serializedStepData as ObjectNode
+        val className = writeableJson.remove("class").asText()
+        return mapper.treeToValue(writeableJson, Class.forName(className)) as T
+    }
 }