Skip to content

Commit 0ccdef5

Browse files
NianJiuZststeipete
NianJiuZst
and
steipete
authored
[codex] Cancel Android gateway pending RPCs on close (#98067)
* fix: cancel Android gateway pending RPCs on close * fix(android): isolate pending RPCs per connection * style(android): separate RPC waiter invariant * chore(android): align native i18n inventory --------- Co-authored-by: NianJiuZst <180004567+users.noreply.github.com> Co-authored-by: Peter Steinberger <[email protected]>
1 parent e3f46d0 commit 0ccdef5

4 files changed

Lines changed: 178 additions & 26 deletions

File tree

‎apps/.i18n/native-source.json‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -259,15 +259,15 @@
259259
},
260260
{
261261
"kind": "conditional-branch",
262-
"line": 1037,
262+
"line": 1054,
263263
"path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
264264
"source": "Connecting…",
265265
"surface": "android",
266266
"id": "native.android.1cbaba9cbffb2f97"
267267
},
268268
{
269269
"kind": "conditional-branch",
270-
"line": 1037,
270+
"line": 1054,
271271
"path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
272272
"source": "Reconnecting…",
273273
"surface": "android",

‎apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt‎

Lines changed: 40 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -180,7 +180,6 @@ class GatewaySession(
180180

181181
private val json = Json { ignoreUnknownKeys = true }
182182
private val writeLock = Mutex()
183-
private val pending = ConcurrentHashMap<String, CompletableDeferred<RpcResponse>>()
184183

185184
@Volatile private var pluginSurfaceUrls: Map<String, String> = emptyMap()
186185

@@ -389,6 +388,11 @@ class GatewaySession(
389388
private var socket: WebSocket? = null
390389
private val loggerTag = "OpenClawGateway"
391390
private val incomingMessages = Channel<String>(Channel.UNLIMITED)
391+
392+
// RPC waiters belong to this socket generation. Closing it must not touch a replacement connection.
393+
private val pending = ConcurrentHashMap<String, CompletableDeferred<RpcResponse>>()
394+
395+
private val pendingLock = Any()
392396
private val messagePumpJob =
393397
scope.launch(Dispatchers.IO) {
394398
for (text in incomingMessages) {
@@ -424,19 +428,14 @@ class GatewaySession(
424428
timeoutMs: Long,
425429
): RpcResponse {
426430
val id = UUID.randomUUID().toString()
427-
val deferred = CompletableDeferred<RpcResponse>()
428-
pending[id] = deferred
431+
val deferred = registerPending(id)
429432
try {
430433
sendJson(buildRequestFrame(id = id, method = method, params = params))
431-
} catch (err: Throwable) {
432-
pending.remove(id)
433-
throw err
434-
}
435-
return try {
436-
withTimeout(timeoutMs) { deferred.await() }
434+
return withTimeout(timeoutMs) { deferred.await() }
437435
} catch (err: TimeoutCancellationException) {
438-
pending.remove(id)
439436
throw IllegalStateException("request timeout")
437+
} finally {
438+
pending.remove(id)
440439
}
441440
}
442441

@@ -447,29 +446,43 @@ class GatewaySession(
447446
onError: (ErrorShape) -> Unit,
448447
) {
449448
val id = UUID.randomUUID().toString()
450-
val deferred = CompletableDeferred<RpcResponse>()
451-
pending[id] = deferred
449+
val deferred = registerPending(id)
452450
try {
453451
sendJson(buildRequestFrame(id = id, method = method, params = params))
454452
} catch (err: Throwable) {
455453
pending.remove(id)
456454
throw err
457455
}
458456
scope.launch(Dispatchers.IO) {
459-
val response =
460-
try {
461-
withTimeout(timeoutMs) { deferred.await() }
462-
} catch (_: TimeoutCancellationException) {
463-
pending.remove(id)
464-
onError(ErrorShape("UNAVAILABLE", "request timeout"))
465-
return@launch
457+
try {
458+
val response =
459+
try {
460+
withTimeout(timeoutMs) { deferred.await() }
461+
} catch (_: TimeoutCancellationException) {
462+
onError(ErrorShape("UNAVAILABLE", "request timeout"))
463+
return@launch
464+
} catch (_: CancellationException) {
465+
return@launch
466+
}
467+
if (!response.ok) {
468+
onError(response.error ?: ErrorShape("UNAVAILABLE", "request failed"))
466469
}
467-
if (!response.ok) {
468-
onError(response.error ?: ErrorShape("UNAVAILABLE", "request failed"))
470+
} finally {
471+
pending.remove(id)
469472
}
470473
}
471474
}
472475

476+
private fun registerPending(id: String): CompletableDeferred<RpcResponse> {
477+
val deferred = CompletableDeferred<RpcResponse>()
478+
// Registration and the close drain are one lifecycle decision; no waiter may slip between them.
479+
synchronized(pendingLock) {
480+
if (isClosed.get()) throw IllegalStateException("Gateway closed")
481+
pending[id] = deferred
482+
}
483+
return deferred
484+
}
485+
473486
suspend fun sendJson(obj: JsonObject) {
474487
val jsonString = obj.toString()
475488
writeLock.withLock {
@@ -500,6 +513,7 @@ class GatewaySession(
500513
if (!connectDeferred.isCompleted) {
501514
connectDeferred.completeExceptionally(IllegalStateException("Gateway closed"))
502515
}
516+
failPending()
503517
socket?.close(1000, "bye")
504518
socket = null
505519
closedDeferred.complete(Unit)
@@ -1011,10 +1025,13 @@ class GatewaySession(
10111025
}
10121026

10131027
private fun failPending() {
1014-
for ((_, waiter) in pending) {
1028+
val waiters =
1029+
synchronized(pendingLock) {
1030+
pending.values.toList().also { pending.clear() }
1031+
}
1032+
for (waiter in waiters) {
10151033
waiter.cancel()
10161034
}
1017-
pending.clear()
10181035
}
10191036
}
10201037

‎apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,14 @@
11
package ai.openclaw.app.gateway
22

3+
import kotlinx.coroutines.CancellationException
34
import kotlinx.coroutines.CompletableDeferred
45
import kotlinx.coroutines.CoroutineScope
56
import kotlinx.coroutines.Dispatchers
67
import kotlinx.coroutines.Job
78
import kotlinx.coroutines.SupervisorJob
89
import kotlinx.coroutines.cancelAndJoin
10+
import kotlinx.coroutines.delay
11+
import kotlinx.coroutines.launch
912
import kotlinx.coroutines.runBlocking
1013
import kotlinx.coroutines.withTimeout
1114
import kotlinx.coroutines.withTimeoutOrNull
@@ -125,6 +128,65 @@ class GatewaySessionInvokeTest {
125128
}
126129
}
127130

131+
@Test
132+
fun disconnectCancelsPendingRpcWithoutWaitingForRequestTimeout() {
133+
runBlocking {
134+
val json = testJson()
135+
val connected = CompletableDeferred<Unit>()
136+
val slowRequestSeen = CompletableDeferred<Unit>()
137+
val requestResult = CompletableDeferred<Result<GatewaySession.RpcResult>>()
138+
val lastDisconnect = AtomicReference("")
139+
val serverWebSocket = AtomicReference<WebSocket?>(null)
140+
val server =
141+
startGatewayServer(json) { webSocket, id, method, _ ->
142+
serverWebSocket.set(webSocket)
143+
when (method) {
144+
"connect" -> webSocket.send(connectResponseFrame(id))
145+
"slow.method" -> {
146+
if (!slowRequestSeen.isCompleted) slowRequestSeen.complete(Unit)
147+
}
148+
}
149+
}
150+
151+
val harness =
152+
createNodeHarness(
153+
connected = connected,
154+
lastDisconnect = lastDisconnect,
155+
) { GatewaySession.InvokeResult.ok("""{"handled":true}""") }
156+
var requestJob: Job? = null
157+
158+
try {
159+
connectNodeSession(harness.session, server.port)
160+
awaitConnectedOrThrow(connected, lastDisconnect, server)
161+
requestJob =
162+
launch {
163+
requestResult.complete(
164+
runCatching {
165+
harness.session.requestDetailed("slow.method", null, timeoutMs = 30_000)
166+
},
167+
)
168+
}
169+
withTimeout(TEST_TIMEOUT_MS) { slowRequestSeen.await() }
170+
171+
harness.session.disconnect()
172+
173+
val result = withTimeout(2_000) { requestResult.await() }
174+
assertEquals(true, result.exceptionOrNull() is CancellationException)
175+
serverWebSocket.get()?.close(1000, "done")
176+
withTimeoutOrNull(2_000) {
177+
while (lastDisconnect.get().isEmpty()) delay(10)
178+
}
179+
} finally {
180+
requestJob?.cancelAndJoin()
181+
runCatching { serverWebSocket.get()?.close(1000, "done") }
182+
delay(100)
183+
harness.session.disconnect()
184+
harness.sessionJob.cancelAndJoin()
185+
server.shutdown()
186+
}
187+
}
188+
}
189+
128190
@Test
129191
fun eventsAreDispatchedInWebSocketFrameOrder() =
130192
runBlocking {

‎apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionReconnectTest.kt‎

Lines changed: 74 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,11 @@ import kotlinx.coroutines.CoroutineScope
55
import kotlinx.coroutines.Dispatchers
66
import kotlinx.coroutines.Job
77
import kotlinx.coroutines.SupervisorJob
8+
import kotlinx.coroutines.async
89
import kotlinx.coroutines.cancelAndJoin
910
import kotlinx.coroutines.runBlocking
1011
import kotlinx.coroutines.withTimeout
12+
import kotlinx.coroutines.withTimeoutOrNull
1113
import kotlinx.serialization.json.Json
1214
import kotlinx.serialization.json.jsonObject
1315
import kotlinx.serialization.json.jsonPrimitive
@@ -28,6 +30,7 @@ import org.robolectric.RobolectricTestRunner
2830
import org.robolectric.RuntimeEnvironment
2931
import org.robolectric.annotation.Config
3032
import java.util.concurrent.ConcurrentLinkedQueue
33+
import java.util.concurrent.atomic.AtomicInteger
3134

3235
private const val LIFECYCLE_TEST_TIMEOUT_MS = 8_000L
3336
private const val LIFECYCLE_CONNECT_CHALLENGE_FRAME =
@@ -79,6 +82,75 @@ private data class ReconnectServer(
7982
@RunWith(RobolectricTestRunner::class)
8083
@Config(sdk = [34])
8184
class GatewaySessionReconnectTest {
85+
@Test
86+
fun staleConnectionDrainCannotCancelReplacementRpc() =
87+
runBlocking {
88+
val json = Json { ignoreUnknownKeys = true }
89+
val firstConnected = CompletableDeferred<Unit>()
90+
val secondConnected = CompletableDeferred<Unit>()
91+
val replacementRequest = CompletableDeferred<Pair<WebSocket, String>>()
92+
val connectionCount = AtomicInteger(0)
93+
val firstServer =
94+
startGatewayServer(json = json) { webSocket, id, method ->
95+
if (method == "connect") webSocket.send(connectResponseFrame(id))
96+
}
97+
val secondServer =
98+
startGatewayServer(json = json) { webSocket, id, method ->
99+
when (method) {
100+
"connect" -> webSocket.send(connectResponseFrame(id))
101+
"slow.method" -> replacementRequest.complete(webSocket to id)
102+
}
103+
}
104+
val harness =
105+
createReconnectHarness(
106+
onConnected = {
107+
when (connectionCount.incrementAndGet()) {
108+
1 -> firstConnected.complete(Unit)
109+
2 -> secondConnected.complete(Unit)
110+
}
111+
},
112+
)
113+
114+
try {
115+
connectNodeSession(harness.session, firstServer.port)
116+
withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { firstConnected.await() }
117+
val oldConnection = readField<Any>(harness.session, "currentConnection")
118+
119+
connectNodeSession(harness.session, secondServer.port)
120+
withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { secondConnected.await() }
121+
val newRequest =
122+
async {
123+
harness.session.requestDetailed("slow.method", null, timeoutMs = 30_000)
124+
}
125+
val (replacementSocket, requestId) =
126+
withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { replacementRequest.await() }
127+
128+
val failPending = oldConnection.javaClass.getDeclaredMethod("failPending")
129+
failPending.isAccessible = true
130+
failPending.invoke(oldConnection)
131+
132+
assertNull(withTimeoutOrNull(200) { newRequest.await() })
133+
replacementSocket.send(
134+
"""{"type":"res","id":"$requestId","ok":true,"payload":{"connection":2}}""",
135+
)
136+
val newResult = withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { newRequest.await() }
137+
assertTrue(newResult.ok)
138+
assertEquals("""{"connection":2}""", newResult.payloadJson)
139+
} finally {
140+
shutdownReconnectHarness(harness, firstServer, secondServer)
141+
}
142+
}
143+
144+
@Suppress("UNCHECKED_CAST")
145+
private fun <T> readField(
146+
target: Any,
147+
name: String,
148+
): T {
149+
val field = target.javaClass.getDeclaredField(name)
150+
field.isAccessible = true
151+
return field.get(target) as T
152+
}
153+
82154
@Test
83155
fun connectToNewGatewayClosesActiveConnectionAndStartsReplacement() =
84156
runBlocking {
@@ -366,6 +438,7 @@ class GatewaySessionReconnectTest {
366438
}
367439

368440
private fun createReconnectHarness(
441+
onConnected: () -> Unit = {},
369442
onConnectFailure: (GatewaySession.ErrorShape, Boolean) -> Unit = { _, _ -> },
370443
): ReconnectHarness {
371444
val app = RuntimeEnvironment.getApplication()
@@ -375,7 +448,7 @@ class GatewaySessionReconnectTest {
375448
scope = CoroutineScope(sessionJob + Dispatchers.Default),
376449
identityStore = DeviceIdentityStore(app),
377450
deviceAuthStore = ReconnectDeviceAuthStore(),
378-
onConnected = {},
451+
onConnected = { onConnected() },
379452
onDisconnected = { _ -> },
380453
onConnectFailure = onConnectFailure,
381454
onEvent = { _, _ -> },

0 commit comments

Comments
 (0)