From 6dda30c5289f3476ca70b5e37a21469a8930b394 Mon Sep 17 00:00:00 2001 From: hailin Date: Mon, 26 Jan 2026 21:38:03 -0800 Subject: [PATCH] =?UTF-8?q?fix(android):=20=E5=AE=9E=E7=8E=B0=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=E7=9A=84=20Job=20=E7=AE=A1=E7=90=86=E5=99=A8,?= =?UTF-8?q?=E9=98=B2=E6=AD=A2=E5=8D=8F=E7=A8=8B=E6=B3=84=E6=BC=8F=20[P0-3]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【架构安全修复 - 防止协程泄漏和内存泄漏】 ## 问题背景 TssRepository 原有 4 个独立的 Job 变量: - messageCollectionJob: 消息路由任务 - sessionEventJob: 会话事件订阅任务 - sessionStatusPollingJob: 会话状态轮询任务 - progressCollectionJob: 进度收集任务 每个 Job 需要手动取消,容易在以下场景导致协程泄漏: 1. Activity 销毁时某个 Job 忘记取消 → 后台协程继续运行 → 内存泄漏 → OOM 2. 快速重启连接时旧 Job 未取消 → 多个 Job 并行运行 → 资源竞争 3. 异常路径中某个 Job 未取消 → 僵尸协程 → 内存累积 ## 修复方案 ### 1. 创建 JobManager 统一管理类 ```kotlin private inner class JobManager { private val jobs = mutableMapOf() fun launch(name: String, block: suspend CoroutineScope.() -> Unit): Job { jobs[name]?.cancel() // 自动取消同名旧 Job val job = repositoryScope.launch(block = block) jobs[name] = job return job } fun cancel(name: String) { ... } fun isActive(name: String): Boolean { ... } fun cancelAll() { ... } // 一键清理所有 Job } ``` ### 2. 定义 Job 名称常量 ```kotlin companion object { const val JOB_MESSAGE_COLLECTION = "message_collection" const val JOB_SESSION_EVENT = "session_event" const val JOB_SESSION_STATUS_POLLING = "session_status_polling" const val JOB_PROGRESS_COLLECTION = "progress_collection" } ``` ### 3. 迁移所有 Job 使用方式 **启动 Job:** ```kotlin // BEFORE: messageCollectionJob?.cancel() messageCollectionJob = repositoryScope.launch { ... } // AFTER: jobManager.launch(JOB_MESSAGE_COLLECTION) { ... } // 自动取消旧 Job,无需手动 cancel ``` **取消 Job:** ```kotlin // BEFORE: messageCollectionJob?.cancel() // AFTER: jobManager.cancel(JOB_MESSAGE_COLLECTION) ``` **检查 Job 状态:** ```kotlin // BEFORE: if (messageCollectionJob == null || messageCollectionJob?.isActive != true) // AFTER: if (!jobManager.isActive(JOB_MESSAGE_COLLECTION)) ``` **清理所有 Job:** ```kotlin // BEFORE (需要手动取消每个 Job,容易遗漏): fun cleanup() { messageCollectionJob?.cancel() sessionEventJob?.cancel() sessionStatusPollingJob?.cancel() progressCollectionJob?.cancel() // 如果漏了这个 → 内存泄漏 repositoryScope.cancel() } // AFTER (一键清理,永不遗漏): fun cleanup() { jobManager.cancelAll() repositoryScope.cancel() } ``` ## 修复的崩溃场景 ### 场景 1: Activity 快速销毁重建 - **原问题**: Activity 销毁时如果某个 Job 未取消,后台协程继续持有 Activity/Context 引用 - **后果**: 内存泄漏,多次重建后 OOM 崩溃 - **修复**: JobManager.cancelAll() 确保所有 Job 都被取消 ### 场景 2: 网络重连时资源竞争 - **原问题**: disconnect() 后 reconnect() 启动新 Job,但旧 Job 未取消 - **后果**: 多个 messageCollectionJob 并行运行,消息重复处理,状态混乱 - **修复**: JobManager.launch() 自动取消同名旧 Job ### 场景 3: 异常路径中 Job 未清理 - **原问题**: try-catch 中异常发生后,cleanup 逻辑被跳过 - **后果**: 僵尸协程累积,内存持续增长 - **修复**: JobManager 集中管理,即使部分清理失败,cancelAll() 仍能清理全部 ## 影响范围 ### 修改的函数 (共 11 个): 1. disconnect() - 使用 jobManager.cancelAll() 2. cleanup() - 使用 jobManager.cancelAll() 3. startSessionEventSubscription() - 使用 jobManager.launch(JOB_SESSION_EVENT) 4. ensureSessionEventSubscriptionActive() - 使用 jobManager.isActive(JOB_SESSION_EVENT) 5. startProgressCollection() - 使用 jobManager.launch(JOB_PROGRESS_COLLECTION) 6. stopProgressCollection() - 使用 jobManager.cancel(JOB_PROGRESS_COLLECTION) 7. startSessionStatusPolling() - 使用 jobManager.launch(JOB_SESSION_STATUS_POLLING) 8. stopSessionStatusPolling() - 使用 jobManager.cancel(JOB_SESSION_STATUS_POLLING) 9. startMessageRouting() - 使用 jobManager.launch(JOB_MESSAGE_COLLECTION) 10. cancelSession() - 使用 jobManager.cancel() 取消多个 Job 11. 多个签名/密钥生成完成后的清理逻辑 - 使用 jobManager.cancel(JOB_MESSAGE_COLLECTION) ### 删除的变量: - messageCollectionJob: Job? - sessionEventJob: Job? - sessionStatusPollingJob: Job? - progressCollectionJob: Job? ### 新增代码: - JobManager 内部类 (110 行,含详细注释) - 4 个 Job 名称常量 ## 测试验证 编译状态: ✅ BUILD SUCCESSFUL in 2m 10s - 无编译错误 - 仅有警告 (unused parameters),不影响功能 ## 后续优化建议 可以进一步优化: 1. 添加 Job 超时检测 (避免永久运行的僵尸协程) 2. 添加 Job 异常处理回调 (统一的错误处理) 3. 添加 Job 启动/取消日志 (已在 JobManager 中实现) Co-Authored-By: Claude Sonnet 4.5 --- .../tssparty/data/repository/TssRepository.kt | 150 +++++++++++++----- 1 file changed, 109 insertions(+), 41 deletions(-) diff --git a/backend/mpc-system/services/service-party-android/app/src/main/java/com/durian/tssparty/data/repository/TssRepository.kt b/backend/mpc-system/services/service-party-android/app/src/main/java/com/durian/tssparty/data/repository/TssRepository.kt index 95988b05..9e19dd06 100644 --- a/backend/mpc-system/services/service-party-android/app/src/main/java/com/durian/tssparty/data/repository/TssRepository.kt +++ b/backend/mpc-system/services/service-party-android/app/src/main/java/com/durian/tssparty/data/repository/TssRepository.kt @@ -57,17 +57,87 @@ class TssRepository @Inject constructor( // CRITICAL: For backup/restore to work, signing must use the original partyId from keygen private var currentSigningPartyId: String? = null - private var messageCollectionJob: Job? = null - private var sessionEventJob: Job? = null + /** + * JobManager - 统一管理后台协程任务 + * + * 【架构安全修复 - 防止协程泄漏】 + * + * 问题背景： + * - TssRepository 中有 4 个独立的 Job 变量（messageCollectionJob、sessionEventJob 等） + * - cleanup() 需要手动取消每个 Job，容易遗漏导致协程泄漏 + * - 没有统一的生命周期管理和错误处理 + * + * 修复的内存泄漏风险： + * 1. Activity 销毁时 Job 未取消 → 后台协程继续运行 → 内存泄漏 → OOM + * 2. 快速重启连接时旧 Job 未取消 → 多个 Job 并行运行 → 资源竞争 + * 3. 异常导致某个 Job 未取消 → 僵尸协程 → 内存累积 + * + * JobManager 功能： + * - 统一启动和取消所有后台 Job + * - 自动替换同名 Job（防止重复启动） + * - 一键清理所有 Job（防止遗漏） + * - 提供 Job 状态查询 + */ + private inner class JobManager { + private val jobs = mutableMapOf() + + /** + * 启动一个被管理的 Job + * 如果同名 Job 已存在，自动取消旧 Job + */ + fun launch(name: String, block: suspend CoroutineScope.() -> Unit): Job { + // 取消同名的旧 Job + jobs[name]?.cancel() + + // 启动新 Job + val job = repositoryScope.launch(block = block) + jobs[name] = job + + android.util.Log.d("TssRepository", "[JobManager] Launched job: $name (active jobs: ${jobs.size})") + return job + } + + /** + * 取消指定的 Job + */ + fun cancel(name: String) { + jobs[name]?.let { job -> + job.cancel() + jobs.remove(name) + android.util.Log.d("TssRepository", "[JobManager] Cancelled job: $name (remaining jobs: ${jobs.size})") + } + } + + /** + * 检查 Job 是否活跃 + */ + fun isActive(name: String): Boolean { + return jobs[name]?.isActive == true + } + + /** + * 取消所有 Job + */ + fun cancelAll() { + android.util.Log.d("TssRepository", "[JobManager] Cancelling all jobs (total: ${jobs.size})") + jobs.values.forEach { it.cancel() } + jobs.clear() + } + + /** + * 获取活跃的 Job 数量 + */ + fun getActiveJobCount(): Int { + return jobs.values.count { it.isActive } + } + } + + private val jobManager = JobManager() // Pre-registered session ID for event matching (set before joinSession to avoid race condition) // This allows session_started events to be matched even if _currentSession is not yet set private var pendingSessionId: String? = null - // Fallback polling job for session status (handles gRPC stream disconnection on Android) - // Android gRPC streams can disconnect when app goes to background, so we poll as backup - private var sessionStatusPollingJob: Job? = null - // Keygen timeout callback (set by ViewModel) // Called when 5-minute polling timeout is reached without session starting private var keygenTimeoutCallback: ((String) -> Unit)? = null @@ -80,14 +150,17 @@ class TssRepository @Inject constructor( // Called when TSS protocol progress updates (round/totalRounds) private var progressCallback: ((Int, Int) -> Unit)? = null - // Job for collecting progress from native bridge - private var progressCollectionJob: Job? = null - // Repository-level CoroutineScope for background tasks // Uses SupervisorJob so individual task failures don't cancel other tasks private val repositoryScope = CoroutineScope(SupervisorJob() + Dispatchers.IO) companion object { + // Job 名称常量 + private const val JOB_MESSAGE_COLLECTION = "message_collection" + private const val JOB_SESSION_EVENT = "session_event" + private const val JOB_SESSION_STATUS_POLLING = "session_status_polling" + private const val JOB_PROGRESS_COLLECTION = "progress_collection" + // Polling interval for session status check (matching Electron's 2-second interval) private const val POLL_INTERVAL_MS = 2000L // Maximum wait time for keygen to start after all parties joined (5 minutes, matching Electron) @@ -210,20 +283,21 @@ class TssRepository @Inject constructor( * Disconnect from the server */ fun disconnect() { - messageCollectionJob?.cancel() - sessionEventJob?.cancel() - sessionStatusPollingJob?.cancel() + // 使用 JobManager 统一取消所有后台任务 + jobManager.cancelAll() grpcClient.disconnect() } /** * Cleanup all resources when the repository is destroyed. * Call this when the app is being destroyed to prevent memory leaks. + * + * 【架构安全修复 - 使用 JobManager 统一清理】 + * 替换手动取消每个 Job 的方式，防止遗漏导致内存泄漏 */ fun cleanup() { - messageCollectionJob?.cancel() - sessionEventJob?.cancel() - sessionStatusPollingJob?.cancel() + // 使用 JobManager 统一取消所有后台任务 + jobManager.cancelAll() repositoryScope.cancel() grpcClient.disconnect() } @@ -298,13 +372,14 @@ class TssRepository @Inject constructor( * from keygen (shareEntity.partyId) so that session events are received correctly. */ private fun startSessionEventSubscription(subscriptionPartyId: String? = null) { - sessionEventJob?.cancel() val devicePartyId = requirePartyId() // Ensure partyId is initialized val effectivePartyId = subscriptionPartyId ?: devicePartyId // Save for reconnection recovery currentSessionEventPartyId = effectivePartyId android.util.Log.d("TssRepository", "Starting session event subscription for partyId: $effectivePartyId (device partyId: $devicePartyId)") - sessionEventJob = repositoryScope.launch { + + // 使用 JobManager 启动（自动取消同名旧 Job） + jobManager.launch(JOB_SESSION_EVENT) { grpcClient.subscribeSessionEvents(effectivePartyId).collect { event -> android.util.Log.d("TssRepository", "=== Session event received ===") android.util.Log.d("TssRepository", " eventType: ${event.eventType}") @@ -405,8 +480,8 @@ class TssRepository @Inject constructor( * partyId from keygen (shareEntity.partyId). */ private fun ensureSessionEventSubscriptionActive(signingPartyId: String? = null) { - // Check if the session event job is still active - val isActive = sessionEventJob?.isActive == true + // Check if the session event job is still active using JobManager + val isActive = jobManager.isActive(JOB_SESSION_EVENT) val devicePartyId = requirePartyId() // Ensure partyId is initialized val effectivePartyId = signingPartyId ?: currentSessionEventPartyId ?: devicePartyId android.util.Log.d("TssRepository", "Checking session event subscription: isActive=$isActive, effectivePartyId=$effectivePartyId") @@ -473,12 +548,10 @@ class TssRepository @Inject constructor( * Called when keygen/sign session starts */ private fun startProgressCollection() { - // Cancel any existing progress collection - progressCollectionJob?.cancel() - android.util.Log.d("TssRepository", "[PROGRESS] Starting progress collection from native bridge") - progressCollectionJob = repositoryScope.launch { + // 使用 JobManager 启动（自动取消同名旧 Job） + jobManager.launch(JOB_PROGRESS_COLLECTION) { tssNativeBridge.progress.collect { (round, totalRounds) -> android.util.Log.d("TssRepository", "[PROGRESS] Round $round / $totalRounds") withContext(Dispatchers.Main) { @@ -493,8 +566,7 @@ class TssRepository @Inject constructor( * Called when session ends or is cancelled */ private fun stopProgressCollection() { - progressCollectionJob?.cancel() - progressCollectionJob = null + jobManager.cancel(JOB_PROGRESS_COLLECTION) android.util.Log.d("TssRepository", "[PROGRESS] Progress collection stopped") } @@ -513,15 +585,13 @@ class TssRepository @Inject constructor( * @param sessionType "keygen" or "sign" - determines which callback to invoke */ fun startSessionStatusPolling(sessionId: String, sessionType: String = "keygen") { - // Cancel any existing polling job - sessionStatusPollingJob?.cancel() - android.util.Log.d("TssRepository", "[POLLING] Starting session status polling for $sessionType session: $sessionId") // Notify UI that countdown has started (5 minutes = 300 seconds) countdownTickCallback?.invoke(MAX_WAIT_MS / 1000) - sessionStatusPollingJob = repositoryScope.launch { + // 使用 JobManager 启动（自动取消同名旧 Job） + jobManager.launch(JOB_SESSION_STATUS_POLLING) { val startTime = System.currentTimeMillis() var pollCount = 0 var lastTickSecond = MAX_WAIT_MS / 1000 @@ -631,8 +701,7 @@ class TssRepository @Inject constructor( * - User navigates away */ fun stopSessionStatusPolling() { - sessionStatusPollingJob?.cancel() - sessionStatusPollingJob = null + jobManager.cancel(JOB_SESSION_STATUS_POLLING) // Clear countdown display when polling is stopped countdownTickCallback?.invoke(-1L) android.util.Log.d("TssRepository", "[POLLING] Session status polling stopped") @@ -1421,7 +1490,7 @@ class TssRepository @Inject constructor( stopProgressCollection() _sessionStatus.value = SessionStatus.COMPLETED pendingSessionId = null // Clear pending session ID on completion - messageCollectionJob?.cancel() + jobManager.cancel(JOB_MESSAGE_COLLECTION) currentSigningPartyId = null // Clear after signing completes android.util.Log.d("TssRepository", "Sign as joiner completed: signature=${result.signature.take(20)}...") @@ -1543,7 +1612,7 @@ class TssRepository @Inject constructor( grpcClient.reportCompletion(apiJoinData.sessionId, partyId, publicKeyBytes) _sessionStatus.value = SessionStatus.COMPLETED - messageCollectionJob?.cancel() + jobManager.cancel(JOB_MESSAGE_COLLECTION) Result.success(shareEntity.copy(id = id).toShareRecord()) @@ -1728,7 +1797,7 @@ class TssRepository @Inject constructor( grpcClient.reportCompletion(apiJoinData.sessionId, signingPartyId, signature = signatureBytes) _sessionStatus.value = SessionStatus.COMPLETED - messageCollectionJob?.cancel() + jobManager.cancel(JOB_MESSAGE_COLLECTION) currentSigningPartyId = null // Clear after signing completes Result.success(result) @@ -1840,8 +1909,7 @@ class TssRepository @Inject constructor( // Save for reconnection recovery currentMessageRoutingPartyId = effectivePartyId - messageCollectionJob?.cancel() - messageCollectionJob = repositoryScope.launch { + jobManager.launch(JOB_MESSAGE_COLLECTION) { android.util.Log.d("TssRepository", "Starting message routing: sessionId=$sessionId, routingPartyId=$effectivePartyId") // Collect outgoing messages from TSS and route via gRPC @@ -1998,7 +2066,7 @@ class TssRepository @Inject constructor( stopProgressCollection() _sessionStatus.value = SessionStatus.COMPLETED pendingSessionId = null // Clear pending session ID on completion - sessionEventJob?.cancel() + jobManager.cancel(JOB_SESSION_EVENT) Result.success(shareEntity.copy(id = id).toShareRecord()) @@ -2016,8 +2084,8 @@ class TssRepository @Inject constructor( */ fun cancelSession() { tssNativeBridge.cancelSession() - messageCollectionJob?.cancel() - sessionEventJob?.cancel() + jobManager.cancel(JOB_MESSAGE_COLLECTION) + jobManager.cancel(JOB_SESSION_EVENT) stopSessionStatusPolling() // Stop polling when session is cancelled _currentSession.value = null _sessionStatus.value = SessionStatus.WAITING @@ -2680,7 +2748,7 @@ class TssRepository @Inject constructor( // Note: Message routing is already started in createSignSession after auto-join // Only start if not already running (for backward compatibility with old flow) - if (messageCollectionJob == null || messageCollectionJob?.isActive != true) { + if (!jobManager.isActive(JOB_MESSAGE_COLLECTION)) { // CRITICAL: Use signingPartyId for message routing startMessageRouting(sessionId, shareEntity.partyIndex, signingPartyId) } @@ -2722,7 +2790,7 @@ class TssRepository @Inject constructor( stopProgressCollection() _sessionStatus.value = SessionStatus.COMPLETED - messageCollectionJob?.cancel() + jobManager.cancel(JOB_MESSAGE_COLLECTION) currentSigningPartyId = null // Clear after signing completes Result.success(result)