merge: W28 —— 进程内共享一条 ChatSocket(连接数 N+2 → 1),SessionsViewModel 为唯一 owner

This commit is contained in:
AskaEth committed 2026-10-05 08:03:50 +08:00
commit 7b97440973
5 files changed
+138 -54

No files matched your search

@@ -202,10 +202,36 @@ class AppContainer(context: Context) {
val personaRepository: PersonaRepository get() = PersonaRepository(personaApi) val personaRepository: PersonaRepository get() = PersonaRepository(personaApi)
fun createChatSocket(): ChatSocket = ChatSocket( /**
baseUrlProvider = { authManager.currentBaseUrl }, * 进程内**唯一**的实时通道连接(`GET /ws/chat`):所有需要收下行帧的 ViewModel 共用这一条。
tokenProvider = authManager::currentToken, *
onAuthRecover = authManager::recoverSocketAuth, * ## 为什么可以共享(不是「省点资源」,是**逐帧等价**)
onAuthFailed = authManager::onSessionExpired, * 服务端是**广播**:下行帧发给**每一条**已认证连接、**不按会话订阅**(见 [ChatSocket] 类注释
) * 「会话归属」)。于是改前「N 条连接,每条都收到全部帧 ⇒ 每个 VM 都收到全部帧 ⇒ 各 VM 靠
* [top.yeij.cyrene.data.ws.WsEvent.isForSession] 收敛到自己的会话」与改后「一条连接收到全部帧
* ⇒ **每个 VM 仍然收到全部帧**」是**同一个语义**:帧集合逐帧相同,过滤与事件分支一个字都不用改。
* 这次变的只是「连接有几条」,不是「帧怎么走」。
*
* ## 为什么必须唯一
* 每个 `ChatSocket` 都自带一个 `OkHttpClient`(各自的连接池 / ping / 重连循环)。改前
* `ChatViewModel`(**每个会话一个 VM**)、[top.yeij.cyrene.ui.sessions.SessionsViewModel]、
* [top.yeij.cyrene.ui.admin.AdminViewModel] 各自建了一条(`ChatViewModel.start()`、
* 两个 VM 的 init)⇒ 访问 N 个会话 = N+2 条连接、N+2 个 `OkHttpClient`;而 VM 活在 Activity 级
* `ViewModelStore` 里,**切走 / 删会话都不回收**。现在这里就是唯一的持有者(`by lazy`,只此一份),
* 调用方只订阅 [ChatSocket.events] / [ChatSocket.state]。
*
* ## 开与关只有一个主人
* [top.yeij.cyrene.ui.sessions.SessionsViewModel] 的登录态订阅:`LoggedIn → connect()`、
* `LoggedOut → close()`(为什么是它,见那个类的 KDoc)。`connect()` 幂等,baseUrl / token 都是
* 每次重连现读 ⇒ 登出重登、换账号**不需要**重建本实例。**任何 ViewModel 都不许关它**——
* 细节与理由写在 [ChatSocket.close]。
*/
val chatSocket: ChatSocket by lazy {
ChatSocket(
baseUrlProvider = { authManager.currentBaseUrl },
tokenProvider = authManager::currentToken,
onAuthRecover = authManager::recoverSocketAuth,
onAuthFailed = authManager::onSessionExpired,
)
}
} }
@@ -32,13 +32,11 @@ import kotlin.random.Random
* Cyrene 实时通道客户端(`GET /ws/chat?token=<JWT>`,api-reference §6)。 * Cyrene 实时通道客户端(`GET /ws/chat?token=<JWT>`,api-reference §6)。
* *
* ## 用法 * ## 用法
* 进程内只有**一个**实例,由 `AppContainer.chatSocket` 持有;上层要收帧就取它,**不要自己 new**
* (见下面「它不是某个屏幕的私有连接」):
* ``` * ```
* val socket = ChatSocket(baseUrlProvider = { cfg.baseUrl }, tokenProvider = { auth.token }, * val socket = container.chatSocket
* onAuthFailed = { auth.logout() }) * // 收集 socket.events,观察 socket.state;发送用 socket.sendChat / socket.sendInterrupt
* socket.connect()
* // 收集 socket.events,观察 socket.state
* socket.sendChat(sessionId = "web", text = "你好", deviceId = deviceId)
* socket.close()
* ``` * ```
* *
* ## 会话归属 * ## 会话归属
@@ -46,6 +44,20 @@ import kotlin.random.Random
* (见 [WsEvent.sessionId]),不补值。服务端广播**不按会话订阅**:同一条连接会收到**所有** * (见 [WsEvent.sessionId]),不补值。服务端广播**不按会话订阅**:同一条连接会收到**所有**
* 会话的帧,因此收集 [events] 的上层**必须**按 [WsEvent.isForSession] 收敛到自己的会话。 * 会话的帧,因此收集 [events] 的上层**必须**按 [WsEvent.isForSession] 收敛到自己的会话。
* *
* ## 它不是某个屏幕的私有连接:谁都不许关它
* 本实例是**进程级共享**的一条连接——所有需要收帧的 ViewModel 订阅的是同一份 [events] / [state]。
* 之所以能共享,正是靠上面那条广播语义:「一条连接收到全部帧」与「N 条连接各收全部帧」对上层是
* **同一个语义**,每个 VM 收到的帧集合逐帧相同,所以共享改的只是「连接有几条」(连带每个连接
* 各带一个 `OkHttpClient` 的成本),不是「帧怎么走」——各 VM 的会话过滤逻辑**一个字都不用动**。
*
* 由此推出两条硬约束:
* 1. **[close] 只归应用级的那条登出 / 会话失效路径**(目前唯一调用点:`SessionsViewModel` 的
* 登录态订阅,见 `AppContainer.chatSocket`)。任何 ViewModel 在 `onCleared` 里关它、页面退出时
* 关它,都会把**别的屏幕**一起掐断——切会话会掐掉抽屉未读与管理页三帧,关管理页会掐掉主会话的
* 流式。连接的生命周期是**进程 + 登录态**,不是屏幕。
* 2. **[events] 的 `replay` 恒为 0**(理由见那里的 KDoc):共享之后,新建的 VM 会立刻订阅同一份流,
* replay > 0 就等于把**历史**下行帧投给一个新来的订阅者。
*
* ## 心跳与重连 * ## 心跳与重连
* - OkHttp `pingInterval` 30s(服务端 ping 周期 54s / pong 等待 60s,客户端 30s 足够)。 * - OkHttp `pingInterval` 30s(服务端 ping 周期 54s / pong 等待 60s,客户端 30s 足够)。
* - 服务端**没有 resume / 补发**:重连成功后由上层用新连接的 `history` 帧 + REST 补差去重。 * - 服务端**没有 resume / 补发**:重连成功后由上层用新连接的 `history` 帧 + REST 补差去重。
@@ -84,7 +96,10 @@ class ChatSocket(
/** /**
* 下行事件流。 * 下行事件流。
* *
* `replay = 0`:重放旧 delta 会让新订阅者重复渲染,会话状态应由 `history` / REST 重建。 * **`replay` 恒为 0,这条不是调优而是必需**:重放旧 delta 会让新订阅者重复渲染,会话状态应由
* `history` / REST 重建;而连接**进程内共享**之后,新建的 VM(切到一个没访问过的会话)会**立刻**
* 订阅同一份流——replay > 0 等于把历史帧投给它,轻则重复渲染,重则会话串台。
*
* `extraBufferCapacity = 512 + DROP_OLDEST`:发送方是 OkHttp 读线程,不能阻塞;溢出时丢最旧的 * `extraBufferCapacity = 512 + DROP_OLDEST`:发送方是 OkHttp 读线程,不能阻塞;溢出时丢最旧的
* 增量帧(服务端发送缓冲 32 本就会丢帧,`stream_end` / `history` 仍可兜底)。 * 增量帧(服务端发送缓冲 32 本就会丢帧,`stream_end` / `history` 仍可兜底)。
*/ */
@@ -114,7 +129,12 @@ class ChatSocket(
private var scope: CoroutineScope? = null private var scope: CoroutineScope? = null
/** 开始连接(幂等:已在连接 / 重连中时忽略)。 */ /**
* 开始连接。**幂等**:已经在连接 / 重连中([wantConnection] 已置位)时直接返回。
*
* 因为幂等,任何一个 VM 都可以在需要时调它做兜底;但**决定它该不该在**的只有登录态那条路径
* (见类注释的硬约束 1),本方法只负责「接上」。
*/
fun connect() { fun connect() {
if (wantConnection) return if (wantConnection) return
wantConnection = true wantConnection = true
@@ -123,7 +143,14 @@ class ChatSocket(
sessionJob = ensureScope().launch { runConnectionLoop() } sessionJob = ensureScope().launch { runConnectionLoop() }
} }
/** 主动断开并停掉重连;之后可再次 [connect]。 */ /**
* 主动断开并停掉重连;之后可再次 [connect]。
*
* ⚠️ 本实例是**进程级共享连接**,不是某个屏幕的私有连接:**只有应用级的登出 / 会话失效路径可以
* 调它**(目前唯一调用点:`SessionsViewModel` 的登录态订阅)。ViewModel 的 `onCleared`、
* 页面退出、切会话都**不许**调——那会把别的屏幕正在用的连接一起掐断。理由见类注释
* 「它不是某个屏幕的私有连接:谁都不许关它」。
*/
fun close() { fun close() {
wantConnection = false wantConnection = false
val socket = webSocket val socket = webSocket
@@ -75,14 +75,18 @@ data class AdminOverviewData(
* 但它**读得到**日志历史([canReadObservabilityLogHistory]),所以概览页上它照样看得到 * 但它**读得到**日志历史([canReadObservabilityLogHistory]),所以概览页上它照样看得到
* 「最近日志」,只是那份列表不增长——这不是错误也不是空态。 * 「最近日志」,只是那份列表不增长——这不是错误也不是空态。
* *
* ## 为什么本类自己开一条 WS 连接 * ## WS 连接:全 App 共享一条,本类**不持有**它
* `observability_*` 三帧在 WS 上,而 `AppContainer` 只提供 `createChatSocket()` * `observability_*` 三帧在 WS 上,而 WS 是**进程内共享**的那一条(`AppContainer.chatSocket`;
* (**每次调用新建**,没有进程级共享连接),`ChatSocket` / 聊天侧却都在别的改动范围里。 * 为什么可以共享见那里的 KDoc——服务端广播 ⇒ 各 VM 各收全部帧 ⇒ 过滤逻辑一个字都不用改)。
* 要在这里收帧,唯一不改别人的办法就是自持一条连接——与 `SessionsViewModel` 同一做法, * 本类只订阅它,**不再自己开一条**。
* 也**同一笔已知技术债**(「每 VM 一条 socket」,收敛方向是「一条应用级共享 socket + *
* 按会话路由」,见 `devdocs/05`;那个重构不在本模块内)。 * **本类任何路径都不 connect / close 它**:
* - 该不该连着由**登录态**决定,主人是 `SessionsViewModel`(它一进登录态就存在,且活在 Activity 级
* `ViewModelStore` 里);本类只在**某个屏幕**出现时才创建,让它插一手就会把「关掉管理页」
* 变成「关掉全 App 的连接」;
* - `viewer` 也不会因此断连:这三帧服务端本就 fail-closed,本类又用 [canReceiveObservabilityFrames]
* 逐帧再挡一次(见 [onEvent]),所以「连接在不在」与「本类收不收到」无关。
* *
* 生命周期:登录且角色够档时连接、登出(或角色降档)断开、VM 销毁断开([onCleared])。
* 本类只关心这三个帧类型,收不到也不会让功能失效——REST 会兜住。 * 本类只关心这三个帧类型,收不到也不会让功能失效——REST 会兜住。
*/ */
class AdminViewModel(private val container: AppContainer) : ViewModel() { class AdminViewModel(private val container: AppContainer) : ViewModel() {
@@ -99,8 +103,11 @@ class AdminViewModel(private val container: AppContainer) : ViewModel() {
private val _backthink = MutableStateFlow(AdminUiState<List<AdminBackthinkItemDto>>()) private val _backthink = MutableStateFlow(AdminUiState<List<AdminBackthinkItemDto>>())
val backthink: StateFlow<AdminUiState<List<AdminBackthinkItemDto>>> = _backthink.asStateFlow() val backthink: StateFlow<AdminUiState<List<AdminBackthinkItemDto>>> = _backthink.asStateFlow()
/** 本 VM 自持的那条推送连接;见类注释。 */ /**
private val socket: ChatSocket = container.createChatSocket() * 全 App 共享的那条推送连接:本类只订阅它,**不连接也不关闭**(理由见类注释——
* 它的主人是「登录态」那条路径,不是某个屏幕)。
*/
private val socket: ChatSocket = container.chatSocket
init { init {
viewModelScope.launch { viewModelScope.launch {
@@ -108,22 +115,20 @@ class AdminViewModel(private val container: AppContainer) : ViewModel() {
when (auth) { when (auth) {
SessionState.Unknown -> Unit SessionState.Unknown -> Unit
// 登出:断开连接并丢掉日志基线(`null` = 没有基线 ⇒ 区块立刻整段收起), // 登出:只丢掉日志基线(`null` = 没有基线 ⇒ 区块立刻整段收起),不留下一个换账号后
// 不留下一个换账号后还挂着上个账号数据的列表。 // 还挂着上个账号数据的列表。**不碰连接**——那是登录态那条路径的事(见类注释),
// 这时主会话的流式与抽屉的未读都还挂着它。
SessionState.LoggedOut -> { SessionState.LoggedOut -> {
socket.close()
_overview.update { state -> _overview.update { state ->
state.copy(data = state.data?.copy(logs = null)) state.copy(data = state.data?.copy(logs = null))
} }
} }
// 连接只在「登录 + 角色够档」时建立:viewer 收不到这三帧(服务端 fail-closed), // 登录:**什么都不做**。连接该不该在由登录态决定(主人是 `SessionsViewModel`);
// 也就没有必要开一条连接去等。角色可能随重登变化,所以这里两个分支都要显式写。 // `viewer` 也不再因此断开——它收不到这三帧是服务端 fail-closed,本类另有
is SessionState.LoggedIn -> if (canReceiveObservabilityFrames(AdminRole.of(auth.role))) { // [canReceiveObservabilityFrames] 逐帧再挡一次(见 [onEvent]),断开连接并不能
socket.connect() // 多挡掉什么,却会掐掉别的屏幕。
} else { is SessionState.LoggedIn -> Unit
socket.close()
}
} }
} }
} }
@@ -131,8 +136,12 @@ class AdminViewModel(private val container: AppContainer) : ViewModel() {
viewModelScope.launch { reloadOnConnect() } viewModelScope.launch { reloadOnConnect() }
} }
/**
* **刻意不关连接**:`socket` 是**全 App 共享**的那一条,而本 VM 只随「离开管理页 / Activity 结束」
* 被销毁——在 `onCleared` 里 close 会把主会话的流式与抽屉的未读一起掐断。
* 断开只由**登录态**表达(见类注释)。
*/
override fun onCleared() { override fun onCleared() {
socket.close()
super.onCleared() super.onCleared()
} }
@@ -516,8 +516,9 @@ data class ChatUiState(
* 上行的两个入口共用这一套实现:输入框的 [send],以及「分享到 App」的 [sendShared] * 上行的两个入口共用这一套实现:输入框的 [send],以及「分享到 App」的 [sendShared]
* (`ACTION_SEND` 进来的内容由 `app/ShareIntake` 拿着,等本会话在登录态下组合出来再落地)。 * (`ACTION_SEND` 进来的内容由 `app/ShareIntake` 拿着,等本会话在登录态下组合出来再落地)。
* *
* 本 VM 只收**属于 [sessionId] 的帧**(见 [onEvent] 的会话闸)。服务端广播不按会话订阅, * 本 VM 只收**属于 [sessionId] 的帧**(见 [onEvent] 的会话闸)。服务端广播不按会话订阅,而连接是
* 且每个会话各有一个 VM(各自的 socket 都还活着),没有这道闸就会串台。 * **进程内共享**的一条(`AppContainer.chatSocket`)⇒ 同一批帧会同时投给**所有**活着的 VM
* (每个会话各有一个 VM),没有这道闸就会串台。
* *
* WS 上行**没有** REST 那样的同步回执:一条帧发出去之后,只有队列满(`pending_full` 广播 + * WS 上行**没有** REST 那样的同步回执:一条帧发出去之后,只有队列满(`pending_full` 广播 +
* 定向 `error{chat_busy}`)与幂等重复(`chat_duplicate`)这两类信号能说明「这条其实没进去 / * 定向 `error{chat_busy}`)与幂等重复(`chat_duplicate`)这两类信号能说明「这条其实没进去 /
@@ -612,11 +613,14 @@ class ChatViewModel(
} }
} }
val ws = container.createChatSocket() // 连接是**进程内共享**的那一条(`AppContainer.chatSocket`),本 VM 只订阅它。开与关归
// 「登录态」那条路径(唯一主人是 `SessionsViewModel`,见它的类注释):`connect()` 虽然幂等,
// 但主人只有一个——本 VM 会随「切到别的会话」被销毁,让它插手就等于让一个屏幕的生死
// 去掐全 App 的连接。
val ws = container.chatSocket
socket = ws socket = ws
viewModelScope.launch { ws.state.collect(::onConnectionChanged) } viewModelScope.launch { ws.state.collect(::onConnectionChanged) }
viewModelScope.launch { ws.events.collect(::onEvent) } viewModelScope.launch { ws.events.collect(::onEvent) }
ws.connect()
} }
/** /**
@@ -1711,9 +1715,12 @@ class ChatViewModel(
*/ */
fun dismissError() = _state.update { it.copy(error = null) } fun dismissError() = _state.update { it.copy(error = null) }
/**
* **刻意不关连接**:`socket` 是**全 App 共享**的那一条(`AppContainer.chatSocket`),而本 VM 随
* 「切到别的会话」就会被销毁。在 `onCleared` 里 close 会连抽屉的未读、管理页的三帧、别的会话的
* 流式一起掐断——连接的开与关只由**登录态**表达(主人是 `SessionsViewModel`,见它的类注释)。
*/
override fun onCleared() { override fun onCleared() {
socket?.close()
socket = null
super.onCleared() super.onCleared()
} }
} }
@@ -69,14 +69,19 @@ data class SessionsUiState(
* 同时调 `POST /sessions/{id}/read` 把游标推上去(策略详见 [pushReadCursor])。 * 同时调 `POST /sessions/{id}/read` 把游标推上去(策略详见 [pushReadCursor])。
* 反向同步(别的端读了)靠下行 `session_read` 帧。 * 反向同步(别的端读了)靠下行 `session_read` 帧。
* *
* ## 为什么本类自己开一条 WS 连接 * ## WS 连接:全 App 共享一条,而**本类是它唯一的主人**
* `session_read` / `message_persisted` 都在 WS 上,而 `AppContainer` 只提供 * `session_read` / `message_persisted` 都在 WS 上,这个连接就是**进程内共享**的那一条
* `createChatSocket()`(**每次调用新建**,没有进程级共享连接),`ChatSocket` / `ChatViewModel` * (`AppContainer.chatSocket`;为什么可以共享见那里的 KDoc——服务端广播 ⇒ 各 VM 各收全部帧
* 又都在别的改动范围里。要在这里收帧,唯一不改别人的办法就是自持一条连接。 * ⇒ 过滤逻辑一个字都不用改)。本类只订阅它,**不再自己开一条**。
* 代价如实说:这是本应用**已知遗留问题**(「每 VM 一条 socket、旧 socket 不关」,见 *
* `devdocs/05`)之上再加一条——服务端对同一账号的多连接是支持的(多端场景本来如此), * **开与关只在这里表达一次**:`LoggedIn → connect()`、`LoggedOut → close()`
* 而且本类只关心两个帧类型,收不到也不会让功能失效(`GET /sessions` 会收敛)。 * ([onCleared] 刻意**不关**,理由见那里)。选本类当主人的两条理由:
* 生命周期:登录后连接、登出断开、VM 销毁断开([onCleared])。 * ① 它由 `ChatHomeScreen` 在**一进登录态**就创建,且活在 Activity 级 `ViewModelStore` 里,存活期
* 正好覆盖「登录 → 登出 → 重登」,与共享连接要的那段时间完全重合;
* ② `ChatViewModel` / `AdminViewModel` 都只在**某个屏幕**出现时才创建(切会话 / 开管理页),让它们
* 当主人就会把「关掉一个屏幕」变成「关掉全 App 的连接」。
*
* baseUrl / token 都是每次重连现读 ⇒ 登出重登、换账号**不需要**重建实例。
*/ */
class SessionsViewModel(private val container: AppContainer) : ViewModel() { class SessionsViewModel(private val container: AppContainer) : ViewModel() {
@@ -98,7 +103,8 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
*/ */
private val readCursors = mutableMapOf<String, Long>() private val readCursors = mutableMapOf<String, Long>()
private val socket: ChatSocket = container.createChatSocket() /** 全 App 共享的那条连接;本类只订阅它,并且**是它唯一的主人**(登录时连、登出时关,见类注释)。 */
private val socket: ChatSocket = container.chatSocket
init { init {
viewModelScope.launch { viewModelScope.launch {
@@ -122,6 +128,10 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
// 登出时把快照一起清掉(先例:EnumCatalogViewModel 在登出时清枚举快照)—— // 登出时把快照一起清掉(先例:EnumCatalogViewModel 在登出时清枚举快照)——
// 否则换账号后第一帧会闪一份上个账号的会话列表。已读游标同理:它是**账号级**的 // 否则换账号后第一帧会闪一份上个账号的会话列表。已读游标同理:它是**账号级**的
// (服务端单一游标),留着会让新账号的会话被旧账号的游标错误地清成已读。 // (服务端单一游标),留着会让新账号的会话被旧账号的游标错误地清成已读。
//
// 这里同时是**全 App 共享连接唯一的那处开与关**(见类注释):本 VM 的存活期正好覆盖
// 「登录 → 登出 → 重登」,而 `ChatViewModel` / `AdminViewModel` 只在某个屏幕出现时才创建,
// 它们不碰连接的生死。
viewModelScope.launch { viewModelScope.launch {
container.authManager.state.collect { auth -> container.authManager.state.collect { auth ->
when (auth) { when (auth) {
@@ -132,9 +142,9 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
_state.value = SessionsUiState() _state.value = SessionsUiState()
} }
// 连接只在登录后建立:Socket 的令牌 / 地址都是每次重连现读的, // 连接只在登录后建立:baseUrl / token 都是每次重连现读的(换账号、续期都不需要
// connect() 幂等(已连接时直接返回)。本 VM 存活期跨越登出 → 重登, // 重建实例),connect() 幂等(已连接时直接返回)。这里订阅的是一份 StateFlow,
// 所以「登出断开、重登接上」必须在这里显式表达。 // 订阅即收到当前值,所以「重登接上」不需要额外的事件。
is SessionState.LoggedIn -> socket.connect() is SessionState.LoggedIn -> socket.connect()
SessionState.Unknown -> Unit SessionState.Unknown -> Unit
@@ -145,8 +155,13 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
viewModelScope.launch { socket.events.collect(::onEvent) } viewModelScope.launch { socket.events.collect(::onEvent) }
} }
/**
* **刻意不关连接**:`socket` 是**全 App 共享**的那一条(`AppContainer.chatSocket`),
* 而本 VM 随 Activity 结束就会被销毁——那时别的屏幕(或下一次进入 App)正要接着用它,
* 在 `onCleared` 里 close 就是「一个 VM 的死掐断全 App 的帧」。断开只由**登录态**表达
* (`LoggedOut` 分支;令牌失效那条路径经 `onAuthFailed` 也收敛到它)。
*/
override fun onCleared() { override fun onCleared() {
socket.close()
super.onCleared() super.onCleared()
} }