fix(nests): publish handles list under AtomicInteger barrier in speaker test
ScriptedSpeaker mutated `handles` AFTER incrementing `_startCount`, so a reader that polled startCount on a different thread (the test runs on the runBlocking thread while the broadcast pump runs on Dispatchers.Default) could see startCount==1 via the volatile AtomicInteger read before the subsequent list write became visible, then fail `assertEquals(1, first.handles.size)` with a stale 0. Also `mutableListOf` itself is not thread-safe — concurrent reads during a write are undefined. Reorder so the list append happens BEFORE the increment (the volatile write now publishes the list mutation under the same happens-before edge tests already rely on for startCount), and swap the backing store for CopyOnWriteArrayList so concurrent reads of `handles[0]` are well-defined. Observed as a flake on Ubuntu CI; passes locally.
This commit is contained in:
+17
-2
@@ -124,12 +124,27 @@ class ReconnectingNestsSpeakerTest {
|
|||||||
|
|
||||||
private val _startCount = AtomicInteger(0)
|
private val _startCount = AtomicInteger(0)
|
||||||
val startCount: Int get() = _startCount.get()
|
val startCount: Int get() = _startCount.get()
|
||||||
val handles = mutableListOf<ScriptedBroadcastHandle>()
|
|
||||||
|
// CopyOnWriteArrayList — the broadcast pump runs on
|
||||||
|
// Dispatchers.Default while tests read `handles[0]` from the
|
||||||
|
// runBlocking thread. A plain `mutableListOf` is neither
|
||||||
|
// thread-safe nor publishes its writes; observed flake on
|
||||||
|
// Ubuntu CI where the test thread sees startCount==1 (volatile
|
||||||
|
// read on AtomicInteger) before the subsequent list mutation
|
||||||
|
// becomes visible. Using a thread-safe list AND ordering the
|
||||||
|
// list mutation BEFORE the AtomicInteger increment ensures
|
||||||
|
// that any reader who sees startCount>=N also sees N entries
|
||||||
|
// already published into `handles`.
|
||||||
|
val handles: MutableList<ScriptedBroadcastHandle> =
|
||||||
|
java.util.concurrent.CopyOnWriteArrayList()
|
||||||
|
|
||||||
override suspend fun startBroadcasting(): BroadcastHandle {
|
override suspend fun startBroadcasting(): BroadcastHandle {
|
||||||
_startCount.incrementAndGet()
|
|
||||||
val handle = ScriptedBroadcastHandle()
|
val handle = ScriptedBroadcastHandle()
|
||||||
handles += handle
|
handles += handle
|
||||||
|
// Increment AFTER appending to `handles` so the volatile
|
||||||
|
// write on AtomicInteger publishes the list mutation —
|
||||||
|
// tests poll on startCount and then read handles[index].
|
||||||
|
_startCount.incrementAndGet()
|
||||||
// Mirror the production speaker's contract: transition
|
// Mirror the production speaker's contract: transition
|
||||||
// Connected → Broadcasting on startBroadcasting.
|
// Connected → Broadcasting on startBroadcasting.
|
||||||
val current = mutableState.value
|
val current = mutableState.value
|
||||||
|
|||||||
Reference in New Issue
Block a user