fix(approval): log + evict on failed approval broadcast (S4)
broadcast() wrapped each client.send in a bare runCatching with no logging or dead-client eviction, so an approval prompt could vanish silently and the session hang until reconnect. Now log+evict failed clients from both registries and error when a prompt reaches 0 live clients. Note: :apps:server:test not run (core:kernel temporarily broken by concurrent work); change is isolated, compiled clean before the breakage, detekt passes.
This commit is contained in:
@@ -13,6 +13,7 @@ import com.correx.core.events.types.SessionId
|
|||||||
import com.correx.core.kernel.orchestration.ApprovalGateway
|
import com.correx.core.kernel.orchestration.ApprovalGateway
|
||||||
import io.ktor.server.websocket.DefaultWebSocketServerSession
|
import io.ktor.server.websocket.DefaultWebSocketServerSession
|
||||||
import io.ktor.websocket.Frame
|
import io.ktor.websocket.Frame
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
import java.util.concurrent.ConcurrentHashMap
|
import java.util.concurrent.ConcurrentHashMap
|
||||||
|
|
||||||
open class ApprovalCoordinator(
|
open class ApprovalCoordinator(
|
||||||
@@ -117,8 +118,29 @@ open class ApprovalCoordinator(
|
|||||||
val encoded = ProtocolSerializer.encodeServerMessage(msg)
|
val encoded = ProtocolSerializer.encodeServerMessage(msg)
|
||||||
val sessionSubs = sessionClients[sessionId].orEmpty()
|
val sessionSubs = sessionClients[sessionId].orEmpty()
|
||||||
val recipients: Set<DefaultWebSocketServerSession> = sessionSubs + globalClients
|
val recipients: Set<DefaultWebSocketServerSession> = sessionSubs + globalClients
|
||||||
|
var delivered = 0
|
||||||
recipients.forEach { client ->
|
recipients.forEach { client ->
|
||||||
runCatching { client.send(Frame.Text(encoded)) }
|
runCatching { client.send(Frame.Text(encoded)) }
|
||||||
|
.onSuccess { delivered++ }
|
||||||
|
.onFailure { err ->
|
||||||
|
// A dead socket must not silently swallow an approval prompt. Log it and evict
|
||||||
|
// the client from both registries so we don't keep sending into a closed socket
|
||||||
|
// (the client re-registers and gets the pending approval from the snapshot on
|
||||||
|
// reconnect).
|
||||||
|
log.warn("approval broadcast to a client failed, evicting: {}", err.message)
|
||||||
|
sessionClients[sessionId]?.remove(client)
|
||||||
|
globalClients.remove(client)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if (delivered == 0) {
|
||||||
|
log.error(
|
||||||
|
"approval prompt for session={} reached 0 live clients; session stays paused until a client reconnects",
|
||||||
|
sessionId.value,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private companion object {
|
||||||
|
private val log = LoggerFactory.getLogger(ApprovalCoordinator::class.java)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user