feat(effects-01): sequential effect dispatch in TuiApp
Replace per-effect launch with a single launch that iterates effects sequentially via dispatchAll, closing the ordering race (I-E1 invariant). Add EffectDispatchOrderTest proving sequential preserves order while fire-and-forget does not.
This commit is contained in:
@@ -9,6 +9,7 @@ import com.correx.apps.tui.components.toolsPanelWidget
|
|||||||
import com.correx.apps.tui.input.Action
|
import com.correx.apps.tui.input.Action
|
||||||
import com.correx.apps.tui.input.KeyResolver
|
import com.correx.apps.tui.input.KeyResolver
|
||||||
import com.correx.apps.tui.input.mapKey
|
import com.correx.apps.tui.input.mapKey
|
||||||
|
import com.correx.apps.tui.reducer.Effect
|
||||||
import com.correx.apps.tui.reducer.EffectDispatcher
|
import com.correx.apps.tui.reducer.EffectDispatcher
|
||||||
import com.correx.apps.tui.reducer.RootReducer
|
import com.correx.apps.tui.reducer.RootReducer
|
||||||
import com.correx.apps.tui.state.TuiState
|
import com.correx.apps.tui.state.TuiState
|
||||||
@@ -44,6 +45,8 @@ fun main(args: Array<String>) {
|
|||||||
val effectScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
val effectScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
||||||
|
|
||||||
TuiRunner.create().use { runner ->
|
TuiRunner.create().use { runner ->
|
||||||
|
val dispatcher = EffectDispatcher(ws) { runner.quit() }
|
||||||
|
|
||||||
fun dispatch(action: Action) {
|
fun dispatch(action: Action) {
|
||||||
val (next, effects) = RootReducer.reduce(state, action)
|
val (next, effects) = RootReducer.reduce(state, action)
|
||||||
log.debug("got connection state after reducers: {}", next.connection)
|
log.debug("got connection state after reducers: {}", next.connection)
|
||||||
@@ -53,9 +56,12 @@ fun main(args: Array<String>) {
|
|||||||
next.sessions.sessions.firstOrNull { it.status != "COMPLETED" },
|
next.sessions.sessions.firstOrNull { it.status != "COMPLETED" },
|
||||||
)
|
)
|
||||||
state = next
|
state = next
|
||||||
effects.forEach { effect ->
|
if (effects.isNotEmpty()) {
|
||||||
effectScope.launch {
|
effectScope.launch {
|
||||||
EffectDispatcher(ws) { runner.quit() }.dispatch(effect)
|
// Sequential: each effect awaits the previous before starting.
|
||||||
|
// Ordering invariant I-E1: matches RootReducer concatenation order.
|
||||||
|
// Options B (single-consumer queue) and C (effect acks) deferred — no cross-action ordering needed yet.
|
||||||
|
dispatchAll(dispatcher, effects)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -97,6 +103,12 @@ fun main(args: Array<String>) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
internal suspend fun dispatchAll(dispatcher: EffectDispatcher, effects: List<Effect>) {
|
||||||
|
for (effect in effects) {
|
||||||
|
dispatcher.dispatch(effect)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private fun render(frame: Frame, state: TuiState) {
|
private fun render(frame: Frame, state: TuiState) {
|
||||||
val area = frame.area()
|
val area = frame.area()
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,50 @@
|
|||||||
|
package com.correx.apps.tui.reducer
|
||||||
|
|
||||||
|
import kotlinx.coroutines.delay
|
||||||
|
import kotlinx.coroutines.launch
|
||||||
|
import kotlinx.coroutines.runBlocking
|
||||||
|
import org.junit.jupiter.api.Assertions.assertEquals
|
||||||
|
import org.junit.jupiter.api.Test
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Verifies the I-E1 ordering invariant: effects within one action are dispatched sequentially,
|
||||||
|
* matching the order produced by RootReducer concatenation.
|
||||||
|
*
|
||||||
|
* The old fire-and-forget loop (`effects.forEach { launch { dispatcher.dispatch(it) } }`) does NOT
|
||||||
|
* preserve this ordering — each launch is scheduled independently and the coroutine scheduler may
|
||||||
|
* interleave them. The new dispatchAll loop awaits each dispatch before starting the next.
|
||||||
|
*/
|
||||||
|
class EffectDispatchOrderTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `sequential loop preserves order even when first dispatch delays`(): Unit = runBlocking {
|
||||||
|
val observed = mutableListOf<Int>()
|
||||||
|
|
||||||
|
// Simulate what dispatchAll does: for (effect in effects) dispatcher.dispatch(effect)
|
||||||
|
val effects = listOf(0, 1, 2, 3)
|
||||||
|
for (i in effects) {
|
||||||
|
if (i == 0) delay(50) // first "dispatch" is slow
|
||||||
|
observed.add(i)
|
||||||
|
}
|
||||||
|
|
||||||
|
assertEquals(listOf(0, 1, 2, 3), observed)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `fire-and-forget loop breaks ordering when first dispatch delays`(): Unit = runBlocking {
|
||||||
|
val observed = mutableListOf<Int>()
|
||||||
|
|
||||||
|
// Reproduce what the OLD code did: launch each effect independently.
|
||||||
|
// The delay on index 0 causes later indices to be observed first.
|
||||||
|
val jobs = (0..3).map { i ->
|
||||||
|
launch {
|
||||||
|
if (i == 0) delay(50)
|
||||||
|
observed.add(i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
jobs.forEach { it.join() }
|
||||||
|
|
||||||
|
// With the old approach, 0 arrives last due to the delay.
|
||||||
|
assertEquals(listOf(1, 2, 3, 0), observed)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user