From fd3c8985ddab59136dba1ee2637c67346b4ad32a Mon Sep 17 00:00:00 2001 From: Alexander Zolotov Date: Mon, 1 Sep 2025 11:41:14 +0200 Subject: [PATCH] [fleet, rpc] restore a guarantee that continuation will never be resumed twice GitOrigin-RevId: 6c72b3758a3a5f79cef0e1f2dfd89f7db7242ce6 --- .../fleet/rpc/client/RpcClient.kt | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/fleet/rpc/srcCommonMain/fleet/rpc/client/RpcClient.kt b/fleet/rpc/srcCommonMain/fleet/rpc/client/RpcClient.kt index 780379b97d99..427724bd9245 100644 --- a/fleet/rpc/srcCommonMain/fleet/rpc/client/RpcClient.kt +++ b/fleet/rpc/srcCommonMain/fleet/rpc/client/RpcClient.kt @@ -412,11 +412,12 @@ private class RpcClient( private fun resumeAllOngoingCallsWithThrowable(throwable: Throwable) { logger.debug(throwable) { "resumeAllOngoingCallsWithThrowable" } - val outgoingRpcIterator = outgoingRpc.iterator() - for ((key, value) in outgoingRpcIterator) { - outgoingRpcIterator.remove() - logger.trace { "resumeAllOngoingCallsWithThrowable: resume request $key" } - value.request.continuation.resumeWithException(throwable) + val outgoingKeys = outgoingRpc.keys.toList() + for (key in outgoingKeys) { + outgoingRpc.remove(key)?.let { + logger.trace { "resumeAllOngoingCallsWithThrowable: resume request $key" } + it.request.continuation.resumeWithException(throwable) + } } streams.values.removeAll { it.closeStream(throwable) @@ -427,11 +428,10 @@ private class RpcClient( private fun resumeWithRouteClosed(route: UID) { val message = "Route $route closed" - val outgoingRpcIterator = outgoingRpc.iterator() - for ((_, value) in outgoingRpcIterator) { - if (value.request.route == route) { - outgoingRpcIterator.remove() - value.request.continuation.resumeWithException(RouteClosedException(route, rpcCallFailureMessage(value.request.call, message))) + val outgoingKeys = outgoingRpc.filterValues { it.request.route == route }.keys + for (key in outgoingKeys) { + outgoingRpc.remove(key)?.let { + it.request.continuation.resumeWithException(RouteClosedException(route, rpcCallFailureMessage(it.request.call, message))) } } streams.values.removeAll {