vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-18 21:21:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(android): order disconnect after in-flight gateway work (#103068)

Browse Source 
* fix(android): drain gateway events before disconnect callback

* chore(android): refresh native i18n inventory

* chore(android): refresh native i18n inventory

---------

Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>
Co-authored-by: Peter Steinberger <peter@steipete.me>
This commit is contained in:
NianJiu
2026-07-10 08:51:51 +08:00
committed by GitHub
parent e6a8feb9c2
commit 05d17289c9
3 changed files with 48 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
apps/.i18n/native-source.json
View File

@@ -403,7 +403,7 @@
},
{
"kind": "conditional-branch",
"line": 1301,
"line": 1303,
"path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
"source": "Connecting…",
"surface": "android",
@@ -411,7 +411,7 @@
},
{
"kind": "conditional-branch",
"line": 1301,
"line": 1303,
"path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
"source": "Reconnecting…",
"surface": "android",

26
apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt
View File

@@ -721,6 +721,7 @@ class GatewaySession(
socket?.cancel() ?: closedDeferred.complete(Unit)
}
@OptIn(DelicateCoroutinesApi::class)
private fun finishTransport(
message: String,
connectError: Throwable,
@@ -728,18 +729,19 @@ class GatewaySession(
if (!terminalCallbackClaimed.compareAndSet(false, true)) return
val shouldNotify = state.getAndSet(ConnectionState.CLOSED) != ConnectionState.CLOSED
incomingMessages.close()
try {
if (shouldNotify) onDisconnected(message)
} finally {
messagePumpJob.invokeOnCompletion {
// OkHttp can deliver onClosed immediately after onMessage. Let an accepted connect
// response finish so auth retry state and issued device tokens survive the close.
if (connectResponseAccepted.get()) {
connectHandshakeJob?.invokeOnCompletion {
finalizeTransport(connectError)
} ?: finalizeTransport(connectError)
} else {
connectNonceDeferred.completeExceptionally(connectError)
// Completion handlers run synchronously and cannot own app-level disconnect cleanup.
connectionScope.launch(Dispatchers.IO, start = CoroutineStart.ATOMIC) {
// Preserve accepted-frame ordering even if the parent scope is cancelled during failure.
withContext(NonCancellable) {
try {
messagePumpJob.join()
if (connectResponseAccepted.get()) {
connectHandshakeJob?.join()
} else {
connectNonceDeferred.completeExceptionally(connectError)
}
if (shouldNotify) onDisconnected(message)
} finally {
finalizeTransport(connectError)
}
}

37
apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionReconnectTest.kt
View File

@@ -127,6 +127,11 @@ private data class ReconnectHarness(
val sessionJob: Job,
)
private data class TerminalCallbackObservation(
val inFlightHandlerCompleted: Boolean,
val issuedTokenPersisted: Boolean,
)
private data class ReconnectServer(
val server: MockWebServer,
val sockets: ConcurrentLinkedQueue<WebSocket>,
@@ -281,14 +286,16 @@ class GatewaySessionReconnectTest {
}
@Test
fun failureDrainsAcceptedConnectResponseBeforeCancellingOwnedWork() =
fun failureOrdersDisconnectAfterInFlightHandlerAndAcceptedConnectResponse() =
runBlocking {
val json = Json { ignoreUnknownKeys = true }
val authStore = RecordingDeviceAuthStore()
val connectRequestId = CompletableDeferred<String>()
val blockEventStarted = CountDownLatch(1)
val allowBlockEvent = CountDownLatch(1)
val terminalCallback = CompletableDeferred<Unit>()
val blockEventCompleted = AtomicBoolean()
val terminalCallback = CompletableDeferred<TerminalCallbackObservation>()
val allowTerminalCallback = CountDownLatch(1)
val retiredInvokeCount = AtomicInteger()
val server =
startGatewayServer(json = json) { _, id, method ->
@@ -297,13 +304,25 @@ class GatewaySessionReconnectTest {
val harness =
createReconnectHarness(
onDisconnected = { message ->
if (message.startsWith("Gateway error:")) terminalCallback.complete(Unit)
if (message.startsWith("Gateway error:")) {
terminalCallback.complete(
TerminalCallbackObservation(
inFlightHandlerCompleted = blockEventCompleted.get(),
issuedTokenPersisted = authStore.savedToken.isCompleted,
),
)
allowTerminalCallback.await(LIFECYCLE_TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS)
}
},
deviceAuthStore = authStore,
onEvent = { event, _ ->
if (event == "block") {
blockEventStarted.countDown()
allowBlockEvent.await(LIFECYCLE_TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS)
try {
allowBlockEvent.await(LIFECYCLE_TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS)
} finally {
blockEventCompleted.set(true)
}
}
},
onInvoke = {
@@ -329,15 +348,23 @@ class GatewaySessionReconnectTest {
"""{"type":"res","id":"$requestId","ok":true,"payload":{"auth":{"deviceToken":"issued-token","role":"node","scopes":[]},"snapshot":{"sessionDefaults":{"mainSessionKey":"main"}}}}""",
)
listener.onFailure(socket, IOException("test failure"), null)
withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { terminalCallback.await() }
assertNull(withTimeoutOrNull(100) { terminalCallback.await() })
allowBlockEvent.countDown()
val observation = withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { terminalCallback.await() }
assertTrue(observation.inFlightHandlerCompleted)
assertTrue(observation.issuedTokenPersisted)
assertEquals("issued-token", withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { authStore.savedToken.await() })
assertEquals(0, retiredInvokeCount.get())
val messagePumpJob = readField<Job>(connection, "messagePumpJob")
assertTrue(withTimeoutOrNull(1_000) { messagePumpJob.join() } != null)
allowTerminalCallback.countDown()
withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { harness.session.disconnectAndJoin() }
} finally {
allowBlockEvent.countDown()
allowTerminalCallback.countDown()
shutdownReconnectHarness(harness, server)
}
}