1093: Launch Side Effects atomically so they are always started

Add headlessIntegrationTest to renderWorkflowIn with a nice Turbine attached.

Check that we do not use the Unconfined dispatcher within the Workflow runtime as it does not provide ordering guarantees.

Update documentation and tests to note the need to use a Dispatcher other than Dispatchers.Unconfined.

Author: steve-the-edwards

benchmarks/performance-poetry/complex-poetry/src/main/java/com/squareup/benchmarks/performance/complex/poetry/PerformancePoetryActivity.kt

@@ -22,7 +22,6 @@ import com.squareup.sample.container.SampleContainers
 import com.squareup.sample.poetry.model.Poem
 import com.squareup.workflow1.RuntimeConfig
 import com.squareup.workflow1.RuntimeConfigOptions.Companion.RENDER_PER_ACTION
-import com.squareup.workflow1.WorkflowExperimentalRuntime
 import com.squareup.workflow1.WorkflowInterceptor
 import com.squareup.workflow1.ui.Screen
 import com.squareup.workflow1.ui.ViewEnvironment.Companion.EMPTY
@@ -31,9 +30,11 @@ import com.squareup.workflow1.ui.WorkflowLayout
 import com.squareup.workflow1.ui.WorkflowUiExperimentalApi
 import com.squareup.workflow1.ui.container.withEnvironment
 import com.squareup.workflow1.ui.renderWorkflowIn
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.plus
 import timber.log.Timber
 
 @OptIn(WorkflowUiExperimentalApi::class)
@@ -53,7 +54,7 @@ class PerformancePoetryActivity : AppCompatActivity() {
   private var selectTimeoutCount = 0
   private var selectTimeoutMainThreadMessageLatch = 0
 
-  @OptIn(WorkflowUiExperimentalApi::class, WorkflowExperimentalRuntime::class)
+  @OptIn(WorkflowUiExperimentalApi::class)
   override fun onCreate(savedInstanceState: Bundle?) {
     super.onCreate(savedInstanceState)
 
@@ -273,7 +274,7 @@ class PoetryModel(
   val renderings: StateFlow<Screen> by lazy {
     renderWorkflowIn(
       workflow = workflow,
-      scope = viewModelScope,
+      scope = viewModelScope + Dispatchers.Main.immediate,
       savedStateHandle = savedState,
       interceptors = interceptor?.let { listOf(it) } ?: emptyList(),
       runtimeConfig = runtimeConfig

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/BaseRenderContext.kt

@@ -30,14 +30,14 @@ import kotlin.reflect.typeOf
  * )
  * ```
  *
- * To create populate such functions from your `render` method, you first need to define a
+ * To create such functions from your `render` method, you first need to define a
  * [WorkflowAction] to handle the event by changing state, emitting an output, or both. Then, just
  * pass a lambda to your rendering that instantiates the action and passes it to
  * [actionSink.send][Sink.send].
  *
  * ## Performing asynchronous work
  *
- * See [runningWorker].
+ * See [runningSideEffect] and [runningWorker].
  *
  * ## Composing children
  *
@@ -92,8 +92,15 @@ public interface BaseRenderContext<out PropsT, StateT, in OutputT> {
    * [cancelled](https://kotlinlang.org/docs/reference/coroutines/cancellation-and-timeouts.html).
    *
    * The coroutine will run with the same [CoroutineContext][kotlin.coroutines.CoroutineContext]
-   * that the workflow runtime is running in. The side effect coroutine will not be started until
-   * _after_ the first render call than runs it returns.
+   * that the workflow runtime is running in.
+   * The coroutine is launched with [CoroutineStart.ATOMIC][kotlinx.coroutines.CoroutineStart.ATOMIC]
+   * start mode, which means that it will _start_ even if the scope is cancelled before it has a
+   * chance to dispatch. This is to guarantee that any time a [sideEffect] is declared running
+   * in any render pass, it will at least be started. If the backing scope is cancelled - it is no
+   * longer declared as running in a consecutive render pass, or the rendering [Workflow] is no
+   * longer rendered - then it will be cancelled at the first suspension point within [sideEffect].
+   *
+   *
    *
    * @param key The string key that is used to distinguish between side effects.
    * @param sideEffect The suspend function that will be launched in a coroutine to perform the

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

@@ -6,12 +6,14 @@ import com.squareup.workflow1.internal.WorkflowRunner
 import com.squareup.workflow1.internal.chained
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
+import kotlin.coroutines.ContinuationInterceptor
 
 /**
  * Launches the [workflow] in a new coroutine in [scope] and returns a [StateFlow] of its
@@ -114,6 +116,14 @@ public fun <PropsT, OutputT, RenderingT> renderWorkflowIn(
 ): StateFlow<RenderingAndSnapshot<RenderingT>> {
   val chainedInterceptor = interceptors.chained()
 
+  check(
+    scope.coroutineContext[ContinuationInterceptor.Key]?.equals(Dispatchers.Unconfined) == false
+  ) {
+    "Cannot use an Unconfined CoroutineDispatcher for the Workflow Runtime. It does not guarantee" +
+      " any dispatch ordering, so we may send actions to unfrozen sinks." +
+      " Use another CoroutineDispatcher. Our recommendation is Main.immediate."
+  }
+
   val runner =
     WorkflowRunner(scope, workflow, props, initialSnapshot, chainedInterceptor, runtimeConfig)
 
@@ -200,7 +210,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,14 @@ 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 ATOMIC because even if the side effect is run and 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
+    // unnecessarily 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)
   }
 }