1093: Launch Side Effects atomically so they are always started

Add headlessIntegrationTest to renderWorkflowIn with a nice Turbine attached.

Still TODO: Add more documentation (and enforcement) of not using Unconfined dispatchers.

Author: steve-the-edwards

gradle.properties

@@ -8,7 +8,7 @@ android.useAndroidX=true
 systemProp.org.gradle.internal.publish.checksums.insecure=true
 
 GROUP=com.squareup.workflow1
-VERSION_NAME=1.11.0-beta04-SNAPSHOT
+VERSION_NAME=1.11.0-beta04-atomic-w-SNAPSHOT
 
 POM_DESCRIPTION=Square Workflow
 

workflow-core/src/commonMain/kotlin/com/squareup/workflow1/Worker.kt

@@ -111,11 +111,6 @@ public interface Worker<out OutputT> {
    * When the worker is torn down, the coroutine is cancelled.
    * This coroutine is launched in the same scope as the workflow runtime, with a few changes:
    *
-   * - The dispatcher is always set to [Unconfined][kotlinx.coroutines.Dispatchers.Unconfined] to
-   *   minimize overhead for workers that don't care which thread they're executed on (e.g. logging
-   *   side effects, workers that wrap third-party reactive libraries, etc.). If your work cares
-   *   which thread it runs on, use [withContext][kotlinx.coroutines.withContext] or
-   *   [flowOn][kotlinx.coroutines.flow.flowOn] to specify a dispatcher.
    * - A [CoroutineName][kotlinx.coroutines.CoroutineName] that describes the `Worker` instance
    *   (via `toString`) and the key specified by the workflow running the worker.
    *

workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/RenderWorkflow.kt

@@ -200,7 +200,7 @@ public fun <PropsT, OutputT, RenderingT> renderWorkflowIn(
         }
       }
 
-      // Pass on to the UI.
+      // Pass the rendering on to the UI.
       renderingsAndSnapshots.value = nextRenderAndSnapshot
       // And emit the Output.
       sendOutput(actionResult, onOutput)

workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowNode.kt

@@ -19,7 +19,7 @@ import com.squareup.workflow1.internal.RealRenderContext.SideEffectRunner
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.CoroutineStart.LAZY
+import kotlinx.coroutines.CoroutineStart.ATOMIC
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
@@ -40,7 +40,10 @@ import kotlin.coroutines.CoroutineContext
  * worker coroutines. This context will override anything from the workflow's scope and any other
  * hard-coded values added to worker contexts. It must not contain a [Job] element (it would violate
  * structured concurrency).
+ *
+ * The opt-in for [ExperimentalCoroutinesApi] is for using a [ATOMIC] on side effect Jobs.
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
   val id: WorkflowNodeId,
   workflow: StatefulWorkflow<PropsT, StateT, OutputT, RenderingT>,
@@ -212,9 +215,7 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
 
     // Tear down workflows and workers that are obsolete.
     subtreeManager.commitRenderedChildren()
-    // Side effect jobs are launched lazily, since they can send actions to the sink, and can only
-    // be started after context is frozen.
-    sideEffects.forEachStaging { it.job.start() }
+    // Tear down side effects that are no longer declared running.
     sideEffects.commitStaging { it.job.cancel() }
 
     return rendering
@@ -260,7 +261,13 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
     sideEffect: suspend CoroutineScope.() -> Unit
   ): SideEffectNode {
     val scope = this + CoroutineName("sideEffect[$key] for $id")
-    val job = scope.launch(start = LAZY, block = sideEffect)
+    // Side effect jobs are automatic because even if the side effect is run, then not run
+    // in consecutive render passes before the side effect can be dispatched, we still want it to
+    // start. Note that this means that side effects must be co-operative or they could unecessarily
+    // hog runtime dispatch. We could force them to be so by adding an if (!isActive) yield() at the
+    // start of the sideEffect block, but that also might mean that expected side effects don't
+    // occur when the sideEffect is run at least once.
+    val job = scope.launch(start = ATOMIC, block = sideEffect)
     return SideEffectNode(key, job)
   }
 }

workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/RenderWorkflowInTest.kt

@@ -9,7 +9,6 @@ import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers.Unconfined
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -27,18 +26,14 @@ import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runCurrent
 import okio.ByteString
+import kotlin.test.Ignore
 import kotlin.test.Test
 
-@OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class, WorkflowExperimentalRuntime::class)
+@OptIn(ExperimentalCoroutinesApi::class, WorkflowExperimentalRuntime::class)
 class RenderWorkflowInTest {
 
   /**
-   * A [TestScope] that will not run until explicitly told to.
-   */
-  private lateinit var pausedTestScope: TestScope
-
-  /**
-   * A [TestScope] that will automatically dispatch enqueued routines.
+   * lateinit so that we can get a fresh scope for each parameterized test run.
    */
   private lateinit var testScope: TestScope
 
@@ -52,8 +47,7 @@ class RenderWorkflowInTest {
   private val runtimeTestRunner = ParameterizedTestRunner<RuntimeConfig>()
 
   private fun setup() {
-    pausedTestScope = TestScope()
-    testScope = TestScope(UnconfinedTestDispatcher())
+    testScope = TestScope()
   }
 
   @Test fun initial_rendering_is_calculated_synchronously() {
@@ -67,7 +61,7 @@ class RenderWorkflowInTest {
 
       val renderings = renderWorkflowIn(
         workflow = workflow,
-        scope = pausedTestScope,
+        scope = testScope,
         props = props,
         runtimeConfig = runtimeConfig
       ) {}
@@ -83,17 +77,18 @@ class RenderWorkflowInTest {
       val props = MutableStateFlow("foo")
       val workflow = Workflow.stateless<String, Nothing, String> { "props: $it" }
 
-      pausedTestScope.cancel()
+      testScope.cancel()
       val renderings = renderWorkflowIn(
         workflow = workflow,
-        scope = pausedTestScope,
+        scope = testScope,
         props = props,
         runtimeConfig = runtimeConfig
       ) {}
       assertEquals("props: foo", renderings.value.rendering)
     }
   }
 
+  @Ignore // Deciding if we want this behavior or if side effects should start.
   @Test
   fun `side_effects_from_initial_rendering_in_root_workflow_are_never_started_when_scope_cancelled_before_start`() {
     runtimeTestRunner.runParametrizedTest(
@@ -120,6 +115,7 @@ class RenderWorkflowInTest {
     }
   }
 
+  @Ignore // Deciding if we want this behavior.
   @Test
   fun `side_effects_from_initial_rendering_in_non_root_workflow_are_never_started_when_scope_cancelled_before_start`() {
     runtimeTestRunner.runParametrizedTest(
@@ -364,12 +360,15 @@ class RenderWorkflowInTest {
       ) {
         receivedOutputs += it
       }
+      testScope.advanceUntilIdle()
       assertTrue(receivedOutputs.isEmpty())
 
       trigger.trySend("foo").isSuccess
+      testScope.advanceUntilIdle()
       assertEquals(listOf("foo"), receivedOutputs)
 
       trigger.trySend("bar").isSuccess
+      testScope.advanceUntilIdle()
       assertEquals(listOf("foo", "bar"), receivedOutputs)
     }
   }
@@ -535,6 +534,7 @@ class RenderWorkflowInTest {
           runtimeConfig = runtimeConfig
         ) {}
       }
+      testScope.advanceUntilIdle()
       assertTrue(sideEffectWasRan)
       assertNotNull(cancellationException)
       val realCause = generateSequence(cancellationException) { it.cause }
@@ -543,6 +543,7 @@ class RenderWorkflowInTest {
     }
   }
 
+  @Ignore // Deciding if we want this behavior, or if all side effects should at least start.
   @Test
   fun `side_effects_from_initial_rendering_in_non_root_workflow_are_never_started_when_initial_render_of_non_root_workflow_fails`() {
     runtimeTestRunner.runParametrizedTest(
@@ -653,10 +654,12 @@ class RenderWorkflowInTest {
         props = MutableStateFlow(Unit),
         runtimeConfig = runtimeConfig
       ) {}
+      testScope.advanceUntilIdle()
       assertNull(cancellationException)
       assertTrue(testScope.isActive)
 
       testScope.cancel()
+      testScope.advanceUntilIdle()
       assertTrue(cancellationException is CancellationException)
       assertNull(cancellationException!!.cause)
     }
@@ -716,10 +719,12 @@ class RenderWorkflowInTest {
         props = MutableStateFlow(Unit),
         runtimeConfig = runtimeConfig
       ) {}
+      testScope.advanceUntilIdle()
       assertNull(cancellationException)
       assertTrue(testScope.isActive)
 
       testScope.cancel(CancellationException("fail!", ExpectedException()))
+      testScope.advanceUntilIdle()
       assertTrue(cancellationException is CancellationException)
       assertTrue(cancellationException!!.cause is ExpectedException)
     }
@@ -766,17 +771,17 @@ class RenderWorkflowInTest {
       }
       val renderings = renderWorkflowIn(
         workflow = workflow,
-        scope = pausedTestScope,
+        scope = testScope,
         props = MutableStateFlow(Unit),
         runtimeConfig = runtimeConfig
       ) {}
 
-      pausedTestScope.launch {
+      testScope.launch {
         renderings.collect { throw ExpectedException() }
       }
       assertNull(cancellationException)
 
-      pausedTestScope.advanceUntilIdle()
+      testScope.advanceUntilIdle()
       assertTrue(cancellationException is CancellationException)
       assertTrue(cancellationException!!.cause is ExpectedException)
     }
@@ -794,20 +799,20 @@ class RenderWorkflowInTest {
       }
       renderWorkflowIn(
         workflow = workflow,
-        scope = pausedTestScope,
+        scope = testScope,
         props = MutableStateFlow(Unit),
         runtimeConfig = runtimeConfig
       ) {
         throw ExpectedException()
       }
-      assertTrue(pausedTestScope.isActive)
+      assertTrue(testScope.isActive)
 
       trigger.complete(Unit)
-      assertTrue(pausedTestScope.isActive)
+      assertTrue(testScope.isActive)
 
-      pausedTestScope.advanceUntilIdle()
-      pausedTestScope.runCurrent()
-      assertFalse(pausedTestScope.isActive)
+      testScope.advanceUntilIdle()
+      testScope.runCurrent()
+      assertFalse(testScope.isActive)
     }
   }
 
@@ -834,20 +839,20 @@ class RenderWorkflowInTest {
 
       renderWorkflowIn(
         workflow = workflow,
-        scope = pausedTestScope,
+        scope = testScope,
         props = MutableStateFlow(Unit),
         runtimeConfig = runtimeConfig,
         onOutput = { events += "output($it)" }
       )
         .onEach { events += "rendering(${it.rendering})" }
-        .launchIn(pausedTestScope)
-      pausedTestScope.advanceUntilIdle()
-      pausedTestScope.runCurrent()
+        .launchIn(testScope)
+      testScope.advanceUntilIdle()
+      testScope.runCurrent()
       assertEquals(listOf("rendering({no output})"), events)
 
       outputTrigger.complete("output")
-      pausedTestScope.advanceUntilIdle()
-      pausedTestScope.runCurrent()
+      testScope.advanceUntilIdle()
+      testScope.runCurrent()
       assertEquals(
         listOf(
           "rendering({no output})",
@@ -974,6 +979,7 @@ class RenderWorkflowInTest {
       val renderings = ras.map { it.rendering }
         .produceIn(testScope)
 
+      testScope.advanceUntilIdle()
       assertFailsWith<ExpectedException> {
         renderings.tryReceive()
           .getOrNull()
@@ -983,7 +989,6 @@ class RenderWorkflowInTest {
 
       props.value += 1
       testScope.advanceUntilIdle()
-      testScope.runCurrent()
 
       assertFailsWith<ExpectedException> {
         renderings.tryReceive()