test(uts): Fix flaky tests - transient-DISCONNECTED CI flake — record-and-verify instead of awaitState

.claude/skills/uts-to-kotlin/references/objects-mapping.md

@@ -566,6 +566,7 @@ message wire forms.**
 | inline ObjectData / map-entry / state fragments | `dataString` / `dataNumber` / `dataBoolean` / `dataObjectId` / `dataBytes` / `dataJson`, `mapEntry`, `mapState`, `counterState`, `mapCreateOp`, `counterCreateOp` |
 | Canonical Constants: `POOL_SERIAL`, `ack_serial(m, i)`, `remote_serial(i)`, `below_ack_serial(i)` | `POOL_SERIAL` (`"t:0"`), `ackSerial(msgSerial, i)`, `remoteSerial(i)`, `belowAckSerial(i)` — use these, never hand-rolled `"t:N"` literals (serials are compared as strings, so ad-hoc values silently sort wrong) |
 | `process_pending_events()` | `` (channel.`object` as DefaultRealtimeObject).asyncFuture { }.await() `` — flushes the objects sequential scope (single-lane FIFO); the empty block cannot run until previously dispatched work has completed or suspended at its wait point |
+| recording lists (`state_changes = []` / `events = []`) | `CopyOnWriteArrayList` — appended on SDK callback threads (objects sequential scope / connection ActionHandler) while `pollUntil`/final asserts read from the test side; never a plain `mutableListOf` |
 
 `mock_ws.send_to_client(...)` is the existing `mockWs.sendToClient(...)` (§ mock API in the main skill). The
 wire `action` / `semantics` are integer enum codes — the builders emit the codes for you.

lib/src/test/kotlin/io/ably/lib/uts/unit/realtime/ConnectionRecoveryTest.kt

@@ -3,6 +3,7 @@ package io.ably.lib.uts.unit.realtime
 import io.ably.lib.uts.infra.unit.*
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.realtime.ConnectionState
+import io.ably.lib.realtime.ConnectionStateListener
 import io.ably.lib.types.ErrorInfo
 import io.ably.lib.types.ProtocolMessage
 import io.ably.lib.types.RecoveryKeyContext
@@ -111,11 +112,19 @@ class ConnectionRecoveryTest {
     assertNotNull(client.connection.createRecoveryKey())
 
     // --- CLOSING and CLOSED states ---
-    // connection.close() sets key = null immediately (Connection.java:116)
+    // connection.close() sets key = null immediately (Connection.java:116). CLOSING is a transient
+    // step on the way to CLOSED, so record connection states BEFORE the close() stimulus and
+    // pollUntil CLOSING is observed rather than awaiting it post-stimulus, which can race that window
+    // (record-and-verify pattern, uts/docs/writing-test-specs.md, "Verifying Transient States").
+    // CLOSED is terminal/sticky, so the existing awaitState suffices for it.
+    val stateChanges = CopyOnWriteArrayList<ConnectionState>()
+    val stateListener = ConnectionStateListener { stateChanges.add(it.current) }
+    client.connection.on(stateListener)
+
     client.connection.close()
     assertNull(client.connection.createRecoveryKey())
 
-    awaitState(client, ConnectionState.closing)
+    pollUntil { ConnectionState.closing in stateChanges }
     assertNull(client.connection.createRecoveryKey())
 
     mock.sendToClientAndClose(ProtocolMessage().apply {

liveobjects/src/test/kotlin/io/ably/lib/liveobjects/uts/proxy/ObjectsFaultsTest.kt

@@ -7,6 +7,7 @@ import io.ably.lib.types.AblyException
 import io.ably.lib.realtime.Channel
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.realtime.ConnectionState
+import io.ably.lib.realtime.ConnectionStateListener
 import io.ably.lib.types.ChannelMode
 import io.ably.lib.types.ChannelOptions
 import io.ably.lib.uts.infra.awaitChannelState
@@ -30,6 +31,7 @@ import org.junit.jupiter.api.BeforeAll
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
 import java.util.UUID
+import java.util.concurrent.CopyOnWriteArrayList
 import kotlin.test.assertEquals
 import kotlin.test.assertIs
 import kotlin.test.assertFailsWith
@@ -85,17 +87,33 @@ class ObjectsFaultsTest {
         )
         val client = proxyClient(session)
         try {
+            val channel = objectChannel(client, channelName)
+
+            // Record connection states BEFORE the disconnect stimulus (channel.attach(), below).
+            // DISCONNECTED is transient (RTN15a reconnects immediately); a post-stimulus awaitState
+            // can miss it, so this listener MUST precede the stimulus — reordering breaks the test.
+            // (see uts/docs/writing-test-specs.md, "Verifying Transient States")
+            val stateChanges = CopyOnWriteArrayList<ConnectionState>()
+            val stateListener = ConnectionStateListener { stateChanges.add(it.current) }
+            client.connection.on(stateListener)
+
             client.connect()
             awaitState(client, ConnectionState.connected, 15.seconds)
 
-            val channel = objectChannel(client, channelName)
-
-            // First attach triggers sync; proxy disconnects mid-sync.
+            // First attach triggers sync; proxy disconnects mid-sync, then the client auto-reconnects.
             channel.attach()
-            awaitState(client, ConnectionState.disconnected, 15.seconds)
-
-            // Client auto-reconnects; re-attach triggers a fresh sync.
+            // Mid-test gate: poll the RECORDED list for the transient DISCONNECTED before the final
+            // awaitState. The proxy only drops the connection after the OBJECT_SYNC round-trips, so at
+            // this point the client is still CONNECTED and awaitState(connected) would no-op — the
+            // assert would then race ahead of the disconnect (observed as [connecting, connected]).
+            pollUntil(30.seconds) { ConnectionState.disconnected in stateChanges }
+            // Final wait targets CONNECTED, a sticky state — safe for awaitState.
             awaitState(client, ConnectionState.connected, 30.seconds)
+            // CONTAINS_IN_ORDER is a subsequence match, so leading initial-connect states are fine.
+            assertContainsInOrder(
+                stateChanges,
+                listOf(ConnectionState.disconnected, ConnectionState.connecting, ConnectionState.connected)
+            )
 
             // get() waits for SYNCED — resolves only if the re-sync completes.
             val root = withRealTimeout(30.seconds) { channel.`object`.get().await() }
@@ -136,14 +154,25 @@ class ObjectsFaultsTest {
             var rootB = withRealTimeout(15.seconds) { channelB.`object`.get().await() }
             pollUntil(10.seconds) { rootB.get("key1").asString().value() == "initial" }
 
+            // Record B's connection states BEFORE the disconnect stimulus (triggerAction, below).
+            // DISCONNECTED is transient (RTN15a reconnects immediately); a post-stimulus awaitState
+            // can miss it, so this listener MUST precede the stimulus — reordering breaks the test.
+            // (see uts/docs/writing-test-specs.md, "Verifying Transient States")
+            val stateChanges = CopyOnWriteArrayList<ConnectionState>()
+            val stateListener = ConnectionStateListener { stateChanges.add(it.current) }
+            clientB.connection.on(stateListener)
+
             // Disconnect client B
             session.triggerAction(mapOf("type" to "disconnect"))
-            awaitState(clientB, ConnectionState.disconnected, 15.seconds)
+            // Mid-test gate: poll the RECORDED list (not awaitState on live state, which could miss
+            // the transient DISCONNECTED). This blocks A's publish until B has observed the drop.
+            pollUntil(15.seconds) { ConnectionState.disconnected in stateChanges }
 
-            // While B is disconnected, A publishes a mutation
+            // A publishes while B is down. Best-effort: RTN15a may reconnect/re-sync B before this
+            // round-trips (then it tests plain delivery, not RTO7/RTO8); the final poll tolerates both.
             rootA.set("key1", LiveMapValue.of("updated_during_disconnect")).await()
 
-            // Client B reconnects and re-syncs; the mutation should be visible
+            // Client B reconnects and re-syncs; the mutation should be visible.
             awaitState(clientB, ConnectionState.connected, 30.seconds)
             rootB = withRealTimeout(15.seconds) { channelB.`object`.get().await() }
             pollUntil(15.seconds) { rootB.get("key1").asString().value() == "updated_during_disconnect" }
@@ -355,4 +384,14 @@ class ObjectsFaultsTest {
                 modes = arrayOf(ChannelMode.object_subscribe, ChannelMode.object_publish)
             },
         )
+
+    /** Asserts [expected] appears in [actual] as an ordered subsequence (the spec's CONTAINS_IN_ORDER). */
+    private fun <T> assertContainsInOrder(actual: List<T>, expected: List<T>) {
+        var i = 0
+        for (item in actual) if (i < expected.size && item == expected[i]) i++
+        assertEquals(
+            expected.size, i,
+            "expected $expected as an ordered subsequence of $actual",
+        )
+    }
 }

liveobjects/src/test/kotlin/io/ably/lib/liveobjects/uts/unit/RealtimeObjectTest.kt

@@ -33,6 +33,7 @@ import io.ably.lib.uts.infra.unit.MockWebSocket
 import io.ably.lib.uts.infra.unit.TestRealtimeClient
 import kotlinx.coroutines.future.await
 import kotlinx.coroutines.test.runTest
+import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -639,7 +640,7 @@ class RealtimeObjectTest {
         val client = newClient(mockWs)
         val channel = client.objectsChannel("test")
 
-        val events = mutableListOf<String>()
+        val events = CopyOnWriteArrayList<String>()
         channel.`object`.on(ObjectStateEvent.SYNCING, ObjectStateChange.Listener { events.add("SYNCING") })
         channel.`object`.on(ObjectStateEvent.SYNCED, ObjectStateChange.Listener { events.add("SYNCED") })
 
@@ -918,8 +919,8 @@ class RealtimeObjectTest {
     fun `RTO24a - RealtimeObject maintains a single PathObjectSubscriptionRegister`() = runTest {
         val (client, _, root, mockWs) = setupSyncedChannel("test")
 
-        val eventsRoot = mutableListOf<PathObjectSubscriptionEvent>()
-        val eventsScore = mutableListOf<PathObjectSubscriptionEvent>()
+        val eventsRoot = CopyOnWriteArrayList<PathObjectSubscriptionEvent>()
+        val eventsScore = CopyOnWriteArrayList<PathObjectSubscriptionEvent>()
 
         // Subscribe via root PathObject at path [].
         root.subscribe(PathObjectListener { event -> eventsRoot.add(event) })
@@ -950,8 +951,8 @@ class RealtimeObjectTest {
     fun `RTO24c1 - subscription coverage prefix match with depth constraint`() = runTest {
         val (client, _, root, mockWs) = setupSyncedChannel("test")
 
-        val shallowEvents = mutableListOf<PathObjectSubscriptionEvent>()
-        val deepEvents = mutableListOf<PathObjectSubscriptionEvent>()
+        val shallowEvents = CopyOnWriteArrayList<PathObjectSubscriptionEvent>()
+        val deepEvents = CopyOnWriteArrayList<PathObjectSubscriptionEvent>()
 
         // Subscribe at root with depth 1 — per RTO24c2b this covers ONLY root's own path ([]),
         // NOT its children (a child like ["score"] is relativeDepth 1-0+1 = 2 > 1).
@@ -1153,7 +1154,7 @@ class RealtimeObjectTest {
     @Test
     fun `RTO20 - subscription fires on apply-on-ACK`() = runTest {
         val (client, _, root, _) = setupSyncedChannel("test")
-        val events = mutableListOf<PathObjectSubscriptionEvent>()
+        val events = CopyOnWriteArrayList<PathObjectSubscriptionEvent>()
         root.get("score").subscribe(PathObjectListener { event -> events.add(event) })
 
         root.get("score").asLiveCounter().increment(10).await()
@@ -1244,7 +1245,10 @@ class RealtimeObjectTest {
             )
             val client = newClient(mockWs)
             val channel = client.objectsChannel("test")
-            val events = mutableListOf<String>()
+            // process_pending_events(): flush the sequential scope so the fresh channel's objects
+            // message pipeline has finished subscribing before attach().
+            (channel.`object` as DefaultRealtimeObject).asyncFuture { }.await()
+            val events = CopyOnWriteArrayList<String>()
             channel.`object`.on(ObjectStateEvent.SYNCING, ObjectStateChange.Listener { events.add("SYNCING") })
             channel.`object`.on(ObjectStateEvent.SYNCED, ObjectStateChange.Listener { events.add("SYNCED") })
 
@@ -1258,7 +1262,7 @@ class RealtimeObjectTest {
         // Scenario "re-sync on new ATTACHED".
         run {
             val (client, channel, _, mockWs) = setupSyncedChannel("test")
-            val events = mutableListOf<String>()
+            val events = CopyOnWriteArrayList<String>()
             channel.`object`.on(ObjectStateEvent.SYNCING, ObjectStateChange.Listener { events.add("SYNCING") })
             channel.`object`.on(ObjectStateEvent.SYNCED, ObjectStateChange.Listener { events.add("SYNCED") })
 
@@ -1275,7 +1279,7 @@ class RealtimeObjectTest {
         // the sync immediately via RTO4b4 → emits SYNCED.
         run {
             val (client, channel, _, mockWs) = setupSyncedChannel("test")
-            val events = mutableListOf<String>()
+            val events = CopyOnWriteArrayList<String>()
             channel.`object`.on(ObjectStateEvent.SYNCING, ObjectStateChange.Listener { events.add("SYNCING") })
             channel.`object`.on(ObjectStateEvent.SYNCED, ObjectStateChange.Listener { events.add("SYNCED") })
 

uts/README.md

@@ -596,6 +596,15 @@ listener — it re-evaluates the predicate every `interval` until it holds or th
 | `awaitChannelState` | `(channel, target, timeout=5s)` | same, for a channel's state |
 | `pollUntil` | `(timeout=15s, interval=100ms) { condition }` | suspend until a boolean predicate holds — used in proxy tests to wait on real network/proxy state, e.g. `pollUntil { authCallbackCount.get() > original }` |
 
+**Transient states and recording lists.** `awaitState`/`awaitChannelState` are for *sticky* targets
+only (CONNECTED, CLOSED, ATTACHED, FAILED…): a transient state — e.g. DISCONNECTED, which RTN15a
+supersedes within microseconds of a drop from CONNECTED — can fire and vanish before any waiter
+registers. Observe transient states with the record-and-verify pattern instead: register a recording
+listener *before* the stimulus, then `pollUntil` on (or assert over) the recorded list. Recording
+lists are appended on SDK callback threads and read from `pollUntil`'s poller thread, so they must be
+thread-safe — always `CopyOnWriteArrayList`, never a plain `mutableListOf` (see the walkthrough in
+§9 and the recording-lists row in the UTS docs' `writing-derived-tests.md`).
+
 A second `Utils.kt` under `infra/unit/` adds the `ConnectionDetails { … }` builder DSL so tests can
 write `ConnectionDetails { connectionKey = "key-1"; connectionStateTtl = 120000L }`. Since this file
 no longer sits in the `io.ably.lib.types` package, it can't call `ConnectionDetails`'s package-private
@@ -655,12 +664,15 @@ One long **await-style** test that walks the SDK through the whole transport lif
    ```
 3. **Publish**, asserting the full MESSAGE frame (`action`, `channel`, `messages[0].name`/`data`) again
    via `awaitNextMessageFromClient()`.
-4. **Disconnect.** `simulateDisconnect()`, await DISCONNECTED, and assert the drop was recorded. Note
-   we do **not** snapshot the `ConnectionAttempt` count here: `FakeClock.waitOn(target, timeout)` does a
-   real `target.wait(timeout)`, so the disconnected-retry fires on its own after ~`disconnectedRetryTimeout`
-   ms of wall-clock even without an `advance()`. `advance()` only wins that race sooner — it is not a
-   hard gate — so a "still exactly one attempt" assertion would be racy on a loaded runner. Ownership
-   of attempt #2 belongs to the next step, which gates on it deterministically.
+4. **Disconnect.** Register a recording list + `ConnectionStateListener` *before* `simulateDisconnect()`,
+   then `pollUntil { disconnected in stateChanges }` — the same inline record-before-stimulus idiom the
+   proxy walkthroughs use (§11.2/§11.3). DISCONNECTED here is **transient**: the drop happens while CONNECTED, so
+   RTN15a reconnects immediately (`Disconnected.enact` queues CONNECTING *before* DISCONNECTED reaches
+   listeners), leaving a microsecond-wide window — independent of `disconnectedRetryTimeout`/`FakeClock`,
+   which never participate on this path. A post-stimulus `awaitState(disconnected)` can race that window
+   and miss it (the CI lost-wakeup flake); recording before the stimulus cannot. We also do **not**
+   snapshot the `ConnectionAttempt` count here — ownership of attempt #2 belongs to the next step, which
+   gates on it deterministically via the buffered `awaitConnectionAttempt()`.
 5. **FakeClock-driven reconnect.** A coroutine loops `fakeClock.advance(2.seconds)` then answers the
    next attempt (received via the buffered `awaitConnectionAttempt()`, so it cannot be missed) with a
    short-TTL CONNECTED; the test awaits CONNECTED again and asserts a second `ConnectionAttempt`.

uts/src/main/kotlin/io/ably/lib/uts/infra/Utils.kt

@@ -17,6 +17,19 @@ import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
 // tryResume/completeResume (the atomic single-winner resume) are @InternalCoroutinesApi.
+/**
+ * Suspends until [client]'s connection reaches [target], or fails with a
+ * [kotlinx.coroutines.TimeoutCancellationException] after [timeout].
+ *
+ * **For sticky targets only** (CONNECTED, CLOSED, SUSPENDED, FAILED, …). It registers its listener
+ * *after* the call point and then checks the current state, so a transition that both arrives and is
+ * superseded before registration is lost. That never happens for a state that persists, but a
+ * **transient** target — DISCONNECTED/CLOSING after a drop, which RTN15a supersedes with CONNECTING
+ * within microseconds — can be missed entirely. For those, use the inline record-before-stimulus
+ * pattern per the UTS record-and-verify convention (spec `uts/docs/writing-test-specs.md`,
+ * "Verifying Transient States"): register a `connection.on { states.add(it.current) }` listener
+ * *before* the stimulus, then `pollUntil { target in states }` (or assert `CONTAINS_IN_ORDER`).
+ */
 @OptIn(InternalCoroutinesApi::class)
 suspend fun awaitState(
   client: AblyRealtime,