Update CSR to yield() and update tests

Author: steve-the-edwards

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

@@ -12,6 +12,7 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.yield
 
 /**
  * Launches the [workflow] in a new coroutine in [scope] and returns a [StateFlow] of its
@@ -170,15 +171,15 @@ public fun <PropsT, OutputT, RenderingT> renderWorkflowIn(
   }
 
   scope.launch {
-    while (isActive) {
+    outer@ while (isActive) {
       // It might look weird to start by processing an action before getting the rendering below,
       // but remember the first render pass already occurred above, before this coroutine was even
       // launched.
       var actionResult: ActionProcessingResult = runner.processAction()
 
       if (shouldShortCircuitForUnchangedState(actionResult)) {
         sendOutput(actionResult, onOutput)
-        continue
+        continue@outer
       }
 
       // After resuming from runner.processAction() our coroutine could now be cancelled, check so
@@ -189,15 +190,23 @@ public fun <PropsT, OutputT, RenderingT> renderWorkflowIn(
       var nextRenderAndSnapshot: RenderingAndSnapshot<RenderingT> = runner.nextRendering()
 
       if (runtimeConfig.contains(CONFLATE_STALE_RENDERINGS)) {
-        while (isActive && actionResult is ActionApplied<*> && actionResult.output == null) {
+        conflate@ while (isActive && actionResult is ActionApplied<*> && actionResult.output == null) {
+          // We start by yielding, because if we are on an Unconfined dispatcher, we want to give
+          // other signals (like Workers listening to the same result) a chance to get dispatched
+          // and queue their actions.
+          yield()
           // We may have more actions we can process, this rendering could be stale.
           actionResult = runner.processAction(waitForAnAction = false)
 
           // If no actions processed, then no new rendering needed. Pass on to UI.
-          if (actionResult == ActionsExhausted) break
+          if (actionResult == ActionsExhausted) break@conflate
 
           // Skip rendering if we had unchanged state, keep draining actions.
-          if (shouldShortCircuitForUnchangedState(actionResult)) continue
+          if (shouldShortCircuitForUnchangedState(actionResult)) {
+            // Emit the Output
+            sendOutput(actionResult, onOutput)
+            continue@outer
+          }
 
           // Make sure the runtime has not been cancelled from runner.processAction()
           if (!isActive) return@launch

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

@@ -23,8 +23,8 @@ import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.advanceUntilIdle
-import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.yield
 import okio.ByteString
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -1343,6 +1343,7 @@ class RenderWorkflowInTest {
         var childHandlerActionExecuted = 0
         var workerActionExecuted = 0
         val trigger = MutableSharedFlow<String>()
+        val outputSet = mutableListOf<String>()
 
         val childWorkflow = Workflow.stateful<String, String, String>(
           initialState = "unchanging state",
@@ -1373,6 +1374,7 @@ class RenderWorkflowInTest {
               action("") {
                 workerActionExecuted++
                 state = it
+                setOutput(it)
               }
             }
             renderState.also {
@@ -1387,7 +1389,9 @@ class RenderWorkflowInTest {
           props = props,
           runtimeConfig = runtimeConfig,
           workflowTracer = workflowTracer,
-        ) {}
+        ) {
+          outputSet.add(it)
+        }
 
         launch {
           trigger.emit("changed state")
@@ -1398,10 +1402,17 @@ class RenderWorkflowInTest {
         assertEquals(2, renderCount)
         assertEquals(1, childHandlerActionExecuted)
         assertEquals(1, workerActionExecuted)
+        assertEquals(1, outputSet.size)
+        assertEquals("changed state", outputSet[0])
       }
     }
   }
 
+  /**
+   * This is the same test as [for_conflate_we_do_not_conflate_stacked_actions_into_one_rendering_if_output]
+   * except that in that version the handler for the child output also sets output - which is
+   * one reason we do not end up conflating.
+   */
   @Test
   fun for_conflate_we_conflate_stacked_actions_into_one_rendering() {
     runtimeTestRunner.runParametrizedTest(
@@ -1411,7 +1422,7 @@ class RenderWorkflowInTest {
         },
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      runTest(StandardTestDispatcher()) {
+      runTest(UnconfinedTestDispatcher()) {
         check(runtimeConfig.contains(CONFLATE_STALE_RENDERINGS))
 
         var childHandlerActionExecuted = false
@@ -1447,6 +1458,7 @@ class RenderWorkflowInTest {
               action("") {
                 // Update the rendering in order to show conflation.
                 state = "$it+update"
+                setOutput(state)
               }
             }
             renderState
@@ -1459,20 +1471,22 @@ class RenderWorkflowInTest {
           props = props,
           runtimeConfig = runtimeConfig,
           workflowTracer = workflowTracer,
-        ) {}
-
-        launch {
-          trigger.emit("changed state")
+        ) {
+          // Yield in output so that we ensure that we let the collector of the renderings
+          // collect each of them before processing the next action.
+          yield()
         }
-        val collectionJob = launch(UnconfinedTestDispatcher(testScheduler)) {
+
+        val collectionJob = launch {
           // Collect this unconfined so we can get all the renderings faster than actions can
           // be processed.
           renderings.collect {
             emitted += it.rendering
           }
         }
-        advanceUntilIdle()
-        runCurrent()
+        launch {
+          trigger.emit("changed state")
+        }
 
         collectionJob.cancel()
 
@@ -1493,7 +1507,7 @@ class RenderWorkflowInTest {
         },
       before = ::setup,
     ) { (runtimeConfig: RuntimeConfig, workflowTracer: WorkflowTracer?) ->
-      runTest(StandardTestDispatcher()) {
+      runTest(UnconfinedTestDispatcher()) {
         check(runtimeConfig.contains(CONFLATE_STALE_RENDERINGS))
 
         var childHandlerActionExecuted = false
@@ -1543,20 +1557,24 @@ class RenderWorkflowInTest {
           props = props,
           runtimeConfig = runtimeConfig,
           workflowTracer = workflowTracer,
-        ) {}
-
-        launch {
-          trigger.emit("changed state")
+        ) {
+          // Yield in output so that we ensure that we let the collector of the renderings
+          // collect each of them before processing the next action.
+          yield()
         }
-        val collectionJob = launch(UnconfinedTestDispatcher(testScheduler)) {
+
+        val collectionJob = launch {
           // Collect this unconfined so we can get all the renderings faster than actions can
           // be processed.
           renderings.collect {
             emitted += it.rendering
           }
         }
+
+        launch {
+          trigger.emit("changed state")
+        }
         advanceUntilIdle()
-        runCurrent()
 
         collectionJob.cancel()
 

workflow-testing/src/test/java/com/squareup/workflow1/WorkflowsLifecycleTests.kt

@@ -4,9 +4,10 @@ import com.squareup.workflow1.RuntimeConfigOptions.CONFLATE_STALE_RENDERINGS
 import com.squareup.workflow1.RuntimeConfigOptions.PARTIAL_TREE_RENDERING
 import com.squareup.workflow1.RuntimeConfigOptions.RENDER_ONLY_WHEN_STATE_CHANGES
 import com.squareup.workflow1.testing.headlessIntegrationTest
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.awaitCancellation
-import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlin.test.Ignore
 import kotlin.test.Test
 
@@ -122,20 +123,45 @@ class WorkflowsLifecycleTests {
     }
   }
 
+  @Test fun childSessionWorkflowStartedWhenExpected() {
+    runtimeTestRunner.runParametrizedTest(
+      paramSource = runtimeOptions,
+      after = ::cleanup,
+    ) { runtimeConfig: RuntimeConfig ->
+
+      workflowWithChildSession.headlessIntegrationTest(
+        runtimeConfig = runtimeConfig
+      ) {
+        // One time starts but does not stop the child session workflow.
+        repeat(1) {
+          val (current, setState) = awaitNextRendering()
+          setState.invoke(current + 1)
+        }
+
+        assertEquals(1, started, "Child Session Workflow not started 1 time.")
+      }
+    }
+  }
+
   /**
    * @see [1093](https://github.com/square/workflow-kotlin/issues/1093)
    *
-   * This test ensconces the currently failing behavior of side effects. We are not currently
-   * fixing this but rather working around it with [SessionWorkflow].
+   * This test fails. It is kept and Ignored as a way to ensconce the currently failing behavior
+   * of side effects with immediate start & stops. We are not currently fixing this but rather
+   * working around it with [SessionWorkflow].
+   *
+   * Compare with [childSessionWorkflowStartAndStoppedWhenHandledSynchronously]
    */
-  @Ignore
-  @Test fun sideEffectsStartAndStoppedWhenHandledSynchronously() {
+  @Ignore("https://github.com/square/workflow-kotlin/issues/1093")
+  @OptIn(ExperimentalCoroutinesApi::class)
+  @Test
+  fun sideEffectsStartAndStoppedWhenHandledSynchronously() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       after = ::cleanup,
     ) { runtimeConfig: RuntimeConfig ->
 
-      val dispatcher = StandardTestDispatcher()
+      val dispatcher = UnconfinedTestDispatcher()
       workflowWithSideEffects.headlessIntegrationTest(
         coroutineContext = dispatcher,
         runtimeConfig = runtimeConfig
@@ -146,9 +172,7 @@ class WorkflowsLifecycleTests {
         // on two consecutive render passes.
         setState.invoke(1)
         setState.invoke(2)
-        dispatcher.scheduler.runCurrent()
         awaitNextRendering()
-        dispatcher.scheduler.runCurrent()
         if (!runtimeConfig.contains(CONFLATE_STALE_RENDERINGS)) {
           // 2 rendering or 1 depending on runtime config.
           awaitNextRendering()
@@ -160,26 +184,6 @@ class WorkflowsLifecycleTests {
     }
   }
 
-  @Test fun childSessionWorkflowStartedWhenExpected() {
-    runtimeTestRunner.runParametrizedTest(
-      paramSource = runtimeOptions,
-      after = ::cleanup,
-    ) { runtimeConfig: RuntimeConfig ->
-
-      workflowWithChildSession.headlessIntegrationTest(
-        runtimeConfig = runtimeConfig
-      ) {
-        // One time starts but does not stop the child session workflow.
-        repeat(1) {
-          val (current, setState) = awaitNextRendering()
-          setState.invoke(current + 1)
-        }
-
-        assertEquals(1, started, "Child Session Workflow not started 1 time.")
-      }
-    }
-  }
-
   @Test fun childSessionWorkflowStoppedWhenExpected() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
@@ -205,13 +209,15 @@ class WorkflowsLifecycleTests {
    *
    * This tests show the working behavior when using a [SessionWorkflow] to track the lifetime.
    */
-  @Test fun childSessionWorkflowStartAndStoppedWhenHandledSynchronously() {
+  @OptIn(ExperimentalCoroutinesApi::class)
+  @Test
+  fun childSessionWorkflowStartAndStoppedWhenHandledSynchronously() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       after = ::cleanup,
     ) { runtimeConfig: RuntimeConfig ->
 
-      val dispatcher = StandardTestDispatcher()
+      val dispatcher = UnconfinedTestDispatcher()
       workflowWithChildSession.headlessIntegrationTest(
         coroutineContext = dispatcher,
         runtimeConfig = runtimeConfig
@@ -222,9 +228,7 @@ class WorkflowsLifecycleTests {
         // on two consecutive render passes, synchronously.
         setState.invoke(1)
         setState.invoke(2)
-        dispatcher.scheduler.runCurrent()
         awaitNextRendering()
-        dispatcher.scheduler.runCurrent()
         if (!runtimeConfig.contains(CONFLATE_STALE_RENDERINGS)) {
           // 2 rendering or 1 depending on runtime config.
           awaitNextRendering()