Add test for events arriving on background thread.

formula/src/main/java/com/instacart/formula/FormulaRuntime.kt

@@ -60,7 +60,13 @@ class FormulaRuntime<Input : Any, Output : Any>(
         this.key = formula.key(input)
 
         if (initialization) {
-            manager = FormulaManagerImpl(this, implementation, input, loggingType = formula::class, inspector = inspector)
+            manager = FormulaManagerImpl(
+                delegate = this,
+                formula = implementation,
+                initialInput = input,
+                loggingType = formula::class,
+                inspector = inspector,
+            )
             run()
 
             hasInitialFinished = true

formula/src/main/java/com/instacart/formula/internal/ChildrenManager.kt

@@ -110,9 +110,9 @@ internal class ChildrenManager(
         val childFormulaHolder = children.findOrInit(key) {
             val implementation = formula.implementation()
             FormulaManagerImpl(
-                delegate,
-                implementation,
-                input,
+                delegate = delegate,
+                formula = implementation,
+                initialInput = input,
                 loggingType = formula::class,
                 inspector = inspector
             )

formula/src/test/java/com/instacart/formula/FormulaRuntimeTest.kt

@@ -3,6 +3,7 @@ package com.instacart.formula
 import com.google.common.truth.Truth
 import com.google.common.truth.Truth.assertThat
 import com.instacart.formula.actions.EmptyAction
+import com.instacart.formula.actions.EventOnBgThreadAction
 import com.instacart.formula.internal.ClearPluginsRule
 import com.instacart.formula.internal.FormulaKey
 import com.instacart.formula.internal.TestInspector
@@ -82,6 +83,7 @@ import org.junit.rules.RuleChain
 import org.junit.rules.TestName
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
+import java.util.concurrent.TimeUnit
 import kotlin.reflect.KClass
 
 @RunWith(Parameterized::class)
@@ -631,6 +633,25 @@ class FormulaRuntimeTest(val runtime: TestableRuntime, val name: String) {
         assertThat(eventCallback.values()).containsExactly("a", "b").inOrder()
     }
 
+    @Test
+    fun `when action returns value on background thread, we emit an error`() {
+        val bgAction = EventOnBgThreadAction()
+        val eventCallback = TestEventCallback<String>()
+        val formula = OnlyUpdateFormula<Unit> {
+            bgAction.onEvent {
+                transition {
+                    eventCallback(it.toString())
+                }
+            }
+        }
+
+        val observer = runtime.test(formula, Unit)
+        bgAction.latch.await(10, TimeUnit.MILLISECONDS)
+        assertThat(bgAction.errors.values().firstOrNull()?.message).contains(
+            "com.instacart.formula.subjects.OnlyUpdateFormula - Only thread that created it can post transition result Expected:"
+        )
+    }
+
     @Test fun `stream is disposed when evaluation does not contain it`() {
         DynamicStreamSubject(runtime)
             .updateStreams(keys = arrayOf("1"))

formula/src/test/java/com/instacart/formula/actions/EventOnBgThreadAction.kt

@@ -0,0 +1,27 @@
+package com.instacart.formula.actions
+
+import com.instacart.formula.Action
+import com.instacart.formula.Cancelable
+import com.instacart.formula.test.TestEventCallback
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+
+class EventOnBgThreadAction : Action<Int> {
+    val errors = TestEventCallback<Throwable>()
+    val latch = CountDownLatch(1)
+
+    override fun key(): Any? = null
+
+    override fun start(send: (Int) -> Unit): Cancelable? {
+        Executors.newSingleThreadExecutor().execute {
+            try {
+                send(0)
+            } catch (e: Throwable) {
+                errors.invoke(e)
+            } finally {
+                latch.countDown()
+            }
+        }
+        return null
+    }
+}