diff --git a/README.md b/README.md index 4f7d68d..2c3a23f 100644 --- a/README.md +++ b/README.md @@ -264,6 +264,9 @@ searchHistoryMaxRecords: 5000 `profileDailyGroupUpdateMaxPendingAgeDays`（默认 7 天）才自动处理，所以旧积压不会被后来的一两条新消息掩盖。 每日任务不使用 20 条的人工全量启动门槛，会把合格群推进到本轮固定历史快照；内容不足门槛的参与者不会调用模型。 它与人工全量共用现有的每群运行锁：仍在推进的群会跳过，已经失败并释放锁的群可以从已保存水位继续处理。 +每日任务内部的模型请求从并发 1 开始，健康请求按 `1 → 4 → 16 → 64 → 配置上限` 渐进放量；任一失败会暂停新请求， +已发出的请求继续完成，失败重试始终保持单请求探测。探测成功后恢复正常准入；连续 3 个不同批次各自耗尽配置的重试 +次数时停止当天任务，未提交批次的水位线保持不变，下一次仍可续跑。该控制器不影响人工画像命令、实时画像或本地回填。 下一次正常群聊会自动携带触发者和最近发言者的认识。现有好感度、Bot 代号、标签和主观印象会与证据驱动的 长期画像按同一个人合并渲染，并明确给出长期画像条目数；私聊也会携带对方的可靠画像摘要和条目数。 diff --git a/src/main/kotlin/profile/ProfileDailyMaintenance.kt b/src/main/kotlin/profile/ProfileDailyMaintenance.kt index 9a1425c..bdf95e3 100644 --- a/src/main/kotlin/profile/ProfileDailyMaintenance.kt +++ b/src/main/kotlin/profile/ProfileDailyMaintenance.kt @@ -107,13 +107,30 @@ object ProfileDailyMaintenance { JChatGPT.logger.info("每日群画像推进开始: groups=${groupIds.size} maxPendingAgeDays=$maxPendingAgeDays") val runToken = UserProfileAnalysisService.newRunToken() + val requestController = ProfileDailyRequestController( + maxConcurrentRequests = PluginConfig.profileMaxConcurrentRequests, + maxAttempts = PluginConfig.profileRetryMax.coerceIn(0, 3) + 1, + ) val outcomes = coroutineScope { groupIds.map { groupId -> - async { analyzeGroup(groupId, runToken) } + async { analyzeGroup(groupId, runToken, requestController) } }.awaitAll() } + val requestStats = requestController.snapshot() val reports = outcomes.mapNotNull(DailyGroupOutcome::report) + outcomes.firstNotNullOfOrNull(DailyGroupOutcome::cause)?.let { cause -> + JChatGPT.logger.error("每日群画像推进存在失败群，本轮仅输出一次代表性异常", cause) + } + if (requestStats.countedFailures > 0) { + val message = "每日群画像请求控制: attempts=${requestStats.totalAttempts} " + + "failures=${requestStats.countedFailures} probes=${requestStats.probeAttempts} " + + "incidents=${requestStats.pauseIncidents} recovered=${requestStats.recoveredIncidents} " + + "exhausted=${requestStats.exhaustedTasks} " + + "peak=${requestStats.peakActive}/${requestStats.peakLimit} " + + "stopped=${requestStats.stopped}" + if (requestStats.stopped) JChatGPT.logger.warning(message) else JChatGPT.logger.info(message) + } JChatGPT.logger.info( "每日群画像推进完成: selected=${groupIds.size} " + "success=${outcomes.count { it.status == DailyGroupStatus.SUCCESS }} " + @@ -132,12 +149,14 @@ object ProfileDailyMaintenance { private suspend fun analyzeGroup( groupId: Long, runToken: ProfileAnalysisRunToken, + requestController: ProfileDailyRequestController, ): DailyGroupOutcome { return try { - val report = UserProfileAnalysisService.analyzeGroup( + val report = UserProfileAnalysisService.analyzeGroupControlled( groupId = groupId, maxBatches = Int.MAX_VALUE, runToken = runToken, + requestController = requestController, ) { progress -> JChatGPT.logger.info( "PROFILE_DAILY_BATCH group=$groupId batch=${progress.batchIndex} " + @@ -155,17 +174,19 @@ object ProfileDailyMaintenance { else -> DailyGroupStatus.SUCCESS } DailyGroupOutcome(status, report.takeUnless { report.alreadyRunning || report.botId == null }) + } catch (cause: ProfileDailyRunStoppedException) { + DailyGroupOutcome(DailyGroupStatus.STOPPED) } catch (cause: CancellationException) { throw cause } catch (cause: Exception) { - JChatGPT.logger.error("群 $groupId 每日画像推进失败", cause) - DailyGroupOutcome(DailyGroupStatus.FAILED) + DailyGroupOutcome(DailyGroupStatus.FAILED, cause = cause) } } private data class DailyGroupOutcome( val status: DailyGroupStatus, val report: GroupProfileAnalysisReport? = null, + val cause: Throwable? = null, ) private enum class DailyGroupStatus { diff --git a/src/main/kotlin/profile/ProfileDailyRequestController.kt b/src/main/kotlin/profile/ProfileDailyRequestController.kt new file mode 100644 index 0000000..d472865 --- /dev/null +++ b/src/main/kotlin/profile/ProfileDailyRequestController.kt @@ -0,0 +1,304 @@ +package top.jie65535.mirai.profile + +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import top.jie65535.mirai.llm.normalizeMaxConcurrentRequests + +/** + * Run-scoped admission controller for the online daily group maintenance job. + * + * The shared ModelService semaphore remains the hard upper bound. This gate + * only controls this one bulk run: it ramps healthy admission from one request, + * pauses fresh work after a failure, and keeps recovery retries single-flight. + */ +internal class ProfileDailyRequestController( + maxConcurrentRequests: Int, + private val maxAttempts: Int, + private val growthFactor: Int = 4, + private val maxConsecutiveExhaustedTasks: Int = 3, +) { + init { + require(maxAttempts > 0) { "maxAttempts must be positive" } + require(growthFactor >= 2) { "growthFactor must be at least 2" } + require(maxConsecutiveExhaustedTasks > 0) { + "maxConsecutiveExhaustedTasks must be positive" + } + } + + private val maxConcurrency = normalizeMaxConcurrentRequests(maxConcurrentRequests) + private val mutex = Mutex() + private val waiters = ArrayDeque() + private val active = LinkedHashMap() + private val heldFailures = LinkedHashSet() + private val exhaustedTasks = LinkedHashSet() + + private var nextPermitId = 1L + private var admissionLimit = 1 + private var successesAtLimit = 0 + private var paused = false + private var incidentOpen = false + private var stopped = false + private var consecutiveExhaustedTasks = 0 + private var freshProbeSuccesses = 0 + + private var totalAttempts = 0 + private var countedFailures = 0 + private var probeAttempts = 0 + private var pauseIncidents = 0 + private var recoveredIncidents = 0 + private var peakActive = 0 + private var peakLimit = 1 + private var maxParallelProbes = 0 + + /** + * Admits and records one model attempt. A rejected/safety response can be + * excluded from the global failure circuit without affecting the gate. + */ + suspend fun execute( + key: ProfileDailyRequestKey, + attempt: Int, + countFailure: (Throwable) -> Boolean = { true }, + block: suspend () -> T, + ): T { + require(attempt in 1..maxAttempts) { "attempt must be within 1..$maxAttempts" } + val permit = awaitPermit(key, attempt) + return try { + val result = block() + onSuccess(permit) + result + } catch (cause: CancellationException) { + onCancellation(permit) + throw cause + } catch (cause: Throwable) { + if (countFailure(cause)) onFailure(permit, attempt) + else release(permit) + throw cause + } + } + + suspend fun snapshot(): ProfileDailyRequestControllerStats = mutex.withLock { + ProfileDailyRequestControllerStats( + totalAttempts = totalAttempts, + countedFailures = countedFailures, + probeAttempts = probeAttempts, + pauseIncidents = pauseIncidents, + recoveredIncidents = recoveredIncidents, + exhaustedTasks = exhaustedTasks.size, + stopped = stopped, + peakActive = peakActive, + peakLimit = peakLimit, + finalLimit = admissionLimit, + maxParallelProbes = maxParallelProbes, + ) + } + + private suspend fun awaitPermit( + key: ProfileDailyRequestKey, + attempt: Int, + ): Permit { + val waiter = Waiter(key, attempt) + mutex.withLock { + if (stopped) { + waiter.deferred.completeExceptionally(ProfileDailyRunStoppedException) + } else { + waiters.addLast(waiter) + pumpLocked() + } + } + return try { + waiter.deferred.await() + } catch (cause: CancellationException) { + mutex.withLock { + waiters.remove(waiter) + waiter.permit?.let { permit -> active.remove(permit.id) } + waiter.deferred.cancel(cause) + pumpLocked() + } + throw cause + } + } + + private suspend fun onSuccess(permit: Permit) { + mutex.withLock { + active.remove(permit.id) + if (permit.kind == PermitKind.RECOVERY) { + heldFailures.remove(permit.key) + if (paused) resumeAfterRecoverySuccessLocked() + } else if (permit.kind == PermitKind.FRESH_PROBE) { + freshProbeSuccesses++ + } + consecutiveExhaustedTasks = 0 + successesAtLimit++ + if (admissionLimit < maxConcurrency && successesAtLimit >= admissionLimit) { + admissionLimit = minOf(maxConcurrency, admissionLimit * growthFactor) + successesAtLimit = 0 + peakLimit = maxOf(peakLimit, admissionLimit) + } + if (paused && heldFailures.isEmpty() && freshProbeSuccesses >= FRESH_PROBE_SUCCESS_TARGET) { + resumeAfterFreshProbesLocked() + } + pumpLocked() + } + } + + private suspend fun onFailure(permit: Permit, attempt: Int) { + mutex.withLock { + active.remove(permit.id) + countedFailures++ + if (stopped) return@withLock + + if (!incidentOpen) { + incidentOpen = true + paused = true + pauseIncidents++ + } + heldFailures.add(permit.key) + if (attempt >= maxAttempts) { + heldFailures.remove(permit.key) + if (exhaustedTasks.add(permit.key)) consecutiveExhaustedTasks++ + if (consecutiveExhaustedTasks >= maxConsecutiveExhaustedTasks) { + stopped = true + failWaitingLocked() + } + } + pumpLocked() + } + } + + private suspend fun onCancellation(permit: Permit) { + mutex.withLock { + active.remove(permit.id) + pumpLocked() + } + } + + private suspend fun release(permit: Permit) { + mutex.withLock { + active.remove(permit.id) + pumpLocked() + } + } + + private fun resumeAfterRecoverySuccessLocked() { + paused = false + if (incidentOpen) { + recoveredIncidents++ + incidentOpen = false + } + freshProbeSuccesses = 0 + } + + private fun resumeAfterFreshProbesLocked() { + paused = false + if (incidentOpen) { + recoveredIncidents++ + incidentOpen = false + } + freshProbeSuccesses = 0 + } + + private fun pumpLocked() { + removeInactiveWaitersLocked() + if (stopped) return + + while (active.size < admissionLimit) { + val waiter = when { + !paused -> findNormalOrRecoveryWaiterLocked() + heldFailures.isNotEmpty() -> { + if (active.values.any { it.kind == PermitKind.RECOVERY }) null + else waiters.firstOrNull { it.attempt > 1 && heldFailures.contains(it.key) } + } + freshProbeSuccesses < FRESH_PROBE_SUCCESS_TARGET -> { + waiters.firstOrNull { it.attempt == 1 } + } + else -> null + } ?: break + + waiters.remove(waiter) + val kind = when { + waiter.attempt > 1 && heldFailures.contains(waiter.key) -> PermitKind.RECOVERY + paused -> PermitKind.FRESH_PROBE + else -> PermitKind.NORMAL + } + val permit = Permit(nextPermitId++, waiter.key, kind) + waiter.permit = permit + active[permit.id] = permit + totalAttempts++ + if (kind != PermitKind.NORMAL) probeAttempts++ + peakActive = maxOf(peakActive, active.size) + maxParallelProbes = maxOf( + maxParallelProbes, + active.values.count { it.kind != PermitKind.NORMAL }, + ) + waiter.deferred.complete(permit) + + if (kind == PermitKind.RECOVERY) break + if (paused) break + } + } + + private fun findNormalOrRecoveryWaiterLocked(): Waiter? { + if (heldFailures.isNotEmpty() && active.values.none { it.kind == PermitKind.RECOVERY }) { + return waiters.firstOrNull { it.attempt > 1 && heldFailures.contains(it.key) } + } + return waiters.firstOrNull { it.attempt == 1 } + } + + private fun removeInactiveWaitersLocked() { + waiters.removeAll { !it.deferred.isActive } + } + + private fun failWaitingLocked() { + waiters.forEach { it.deferred.completeExceptionally(ProfileDailyRunStoppedException) } + waiters.clear() + } + + private data class Waiter( + val key: ProfileDailyRequestKey, + val attempt: Int, + val deferred: CompletableDeferred = CompletableDeferred(), + var permit: Permit? = null, + ) + + private data class Permit( + val id: Long, + val key: ProfileDailyRequestKey, + val kind: PermitKind, + ) + + private enum class PermitKind { + NORMAL, + RECOVERY, + FRESH_PROBE, + } + + companion object { + private const val FRESH_PROBE_SUCCESS_TARGET = 2 + } +} + +internal data class ProfileDailyRequestKey( + val groupId: Long, + val startTime: Int, + val endTime: Int, +) + +internal data class ProfileDailyRequestControllerStats( + val totalAttempts: Int, + val countedFailures: Int, + val probeAttempts: Int, + val pauseIncidents: Int, + val recoveredIncidents: Int, + val exhaustedTasks: Int, + val stopped: Boolean, + val peakActive: Int, + val peakLimit: Int, + val finalLimit: Int, + val maxParallelProbes: Int, +) + +internal object ProfileDailyRunStoppedException : RuntimeException( + "每日群画像推进已因连续模型请求失败而停止", +) diff --git a/src/main/kotlin/profile/UserProfileAnalysisService.kt b/src/main/kotlin/profile/UserProfileAnalysisService.kt index 8acf830..d04abd0 100644 --- a/src/main/kotlin/profile/UserProfileAnalysisService.kt +++ b/src/main/kotlin/profile/UserProfileAnalysisService.kt @@ -170,6 +170,34 @@ object UserProfileAnalysisService { maxBatches: Int, runToken: ProfileAnalysisRunToken = newRunToken(), onProgress: suspend (GroupProfileAnalysisProgress) -> Unit = {}, + ): GroupProfileAnalysisReport = analyzeGroupInternal( + groupId = groupId, + maxBatches = maxBatches, + runToken = runToken, + requestController = null, + onProgress = onProgress, + ) + + internal suspend fun analyzeGroupControlled( + groupId: Long, + maxBatches: Int, + runToken: ProfileAnalysisRunToken, + requestController: ProfileDailyRequestController, + onProgress: suspend (GroupProfileAnalysisProgress) -> Unit = {}, + ): GroupProfileAnalysisReport = analyzeGroupInternal( + groupId = groupId, + maxBatches = maxBatches, + runToken = runToken, + requestController = requestController, + onProgress = onProgress, + ) + + private suspend fun analyzeGroupInternal( + groupId: Long, + maxBatches: Int, + runToken: ProfileAnalysisRunToken, + requestController: ProfileDailyRequestController?, + onProgress: suspend (GroupProfileAnalysisProgress) -> Unit, ): GroupProfileAnalysisReport { require(groupId > 0) { "groupId 必须是正数" } require(maxBatches > 0) { "maxBatches 必须是正数" } @@ -194,7 +222,7 @@ object UserProfileAnalysisService { } try { - return analyzeGroupExclusive(groupId, maxBatches, runToken, onProgress) + return analyzeGroupExclusive(groupId, maxBatches, runToken, requestController, onProgress) } finally { runningGroups.remove(groupId) } @@ -306,6 +334,7 @@ object UserProfileAnalysisService { groupId: Long, maxBatches: Int, runToken: ProfileAnalysisRunToken, + requestController: ProfileDailyRequestController?, onProgress: suspend (GroupProfileAnalysisProgress) -> Unit, ): GroupProfileAnalysisReport { val reader = withContext(Dispatchers.IO) { ProfileHistoryReader(resolveHistoryFile()) } @@ -367,6 +396,7 @@ object UserProfileAnalysisService { summaryMaxLength = PluginConfig.profileSummaryMaxLength, onRetryFailure = { message, cause -> JChatGPT.logger.warning(message, cause) }, onCommittedOperations = ProfileOperationLogger::log, + requestController = requestController, ) } catch (cause: ModelSafetyRejectionException) { JChatGPT.logger.warning( @@ -461,6 +491,7 @@ object UserProfileAnalysisService { summaryMaxLength: Int, onRetryFailure: (String, Throwable) -> Unit = { _, _ -> }, onCommittedOperations: (String, Collection) -> Unit = { _, _ -> }, + requestController: ProfileDailyRequestController? = null, ): ConversationProfileAnalysisReport? { check(UserProfileStore.isAvailable) { "用户画像数据库不可用" } val eligibleUserIds = batch.authoredTextCharsByUser @@ -486,6 +517,7 @@ object UserProfileAnalysisService { retryMax = retryMax, summaryMaxLength = summaryMaxLength, onRetryFailure = onRetryFailure, + requestController = requestController, ) val totalUsage = result.usage val commitOutcome = userLocks.withUserLocks(eligibleUserIds) { @@ -569,36 +601,57 @@ object UserProfileAnalysisService { retryMax: Int, summaryMaxLength: Int, onRetryFailure: (String, Throwable) -> Unit, + requestController: ProfileDailyRequestController?, ): Pair> { val attempts = retryMax.coerceIn(0, 3) + 1 val retryBackoff = RetryBackoff.fromConfig() + val requestKey = ProfileDailyRequestKey( + groupId = batch.groupId, + startTime = batch.startTime, + endTime = batch.endTime, + ) var lastFailure: Throwable? = null repeat(attempts) { attempt -> try { - val result = model.analyzeConversation( - profiles, - batch, - eligibleUserIds, - supportStatsByUserId, - ) - val reductions = ConversationProfileReducer.reduce( - profiles = profiles, - batch = batch, - eligibleUserIds = eligibleUserIds, - response = result.response, - model = model.modelName, - promptVersion = ProfilePromptStore.PROMPT_VERSION, - summaryMaxLength = summaryMaxLength.coerceAtLeast(100), - ) - return result to reductions + val analyzeAttempt: suspend () -> Pair> = { + val result = model.analyzeConversation( + profiles, + batch, + eligibleUserIds, + supportStatsByUserId, + ) + val reductions = ConversationProfileReducer.reduce( + profiles = profiles, + batch = batch, + eligibleUserIds = eligibleUserIds, + response = result.response, + model = model.modelName, + promptVersion = ProfilePromptStore.PROMPT_VERSION, + summaryMaxLength = summaryMaxLength.coerceAtLeast(100), + ) + result to reductions + } + return if (requestController == null) { + analyzeAttempt() + } else { + requestController.execute( + key = requestKey, + attempt = attempt + 1, + countFailure = { cause -> cause !is ModelRequestRejectedException }, + block = analyzeAttempt, + ) + } } catch (cause: Exception) { + if (cause is ProfileDailyRunStoppedException) throw cause if (cause is CancellationException) throw cause if (cause is ModelRequestRejectedException) { - onRetryFailure( - "群 ${batch.groupId} 会话画像 [${batch.startTime}, ${batch.endTime}) " + - "被模型拒绝，已停止重试", - cause, - ) + if (requestController == null) { + onRetryFailure( + "群 ${batch.groupId} 会话画像 [${batch.startTime}, ${batch.endTime}) " + + "被模型拒绝，已停止重试", + cause, + ) + } throw cause } lastFailure = cause @@ -609,7 +662,7 @@ object UserProfileAnalysisService { "第 ${attempt + 1}/$attempts 次分析失败", cause = cause, retryBackoff = retryBackoff, - logFailure = onRetryFailure, + logFailure = if (requestController == null) onRetryFailure else { _, _ -> }, ) } } diff --git a/src/test/kotlin/profile/ProfileDailyRequestControllerTest.kt b/src/test/kotlin/profile/ProfileDailyRequestControllerTest.kt new file mode 100644 index 0000000..d66933b --- /dev/null +++ b/src/test/kotlin/profile/ProfileDailyRequestControllerTest.kt @@ -0,0 +1,258 @@ +package top.jie65535.mirai.profile + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.withTimeoutOrNull +import java.io.IOException +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertFalse +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class ProfileDailyRequestControllerTest { + @Test + fun growsHealthyAdmissionByFourWithoutExceedingTheConfiguredCap() = runBlocking { + val controller = controller(maxConcurrency = 128) + + succeed(controller, key(1)) + assertEquals(4, controller.snapshot().finalLimit) + + repeat(4) { index -> succeed(controller, key(index + 2)) } + assertEquals(16, controller.snapshot().finalLimit) + + repeat(16) { index -> succeed(controller, key(index + 6)) } + assertEquals(64, controller.snapshot().finalLimit) + + repeat(64) { index -> succeed(controller, key(index + 22)) } + val stats = controller.snapshot() + assertEquals(128, stats.finalLimit) + assertEquals(128, stats.peakLimit) + assertFalse(stats.stopped) + } + + @Test + fun pausesFreshAdmissionUntilTheFailedTaskRecovers() = runBlocking { + val controller = controller(maxConcurrency = 4) + succeed(controller, key(1)) + val failedKey = key(2) + fail(controller, failedKey, attempt = 1) + val freshStarted = CompletableDeferred() + + coroutineScope { + val fresh = async { + controller.execute(key(3), attempt = 1) { + freshStarted.complete(Unit) + } + } + assertNull(withTimeoutOrNull(50) { freshStarted.await() }) + + succeed(controller, failedKey, attempt = 2) + withTimeout(1_000) { freshStarted.await() } + fresh.await() + } + + val stats = controller.snapshot() + assertEquals(1, stats.countedFailures) + assertEquals(1, stats.recoveredIncidents) + assertEquals(1, stats.maxParallelProbes) + assertFalse(stats.stopped) + } + + @Test + fun admitsNewWorkWhenOneFastRequestFinishesBeforeSlowRequests() = runBlocking { + val controller = controller(maxConcurrency = 4) + succeed(controller, key(1)) + val allEntered = CompletableDeferred() + val releaseFast = CompletableDeferred() + val releaseSlow = CompletableDeferred() + val entered = AtomicInteger() + + coroutineScope { + val activeRequests = (2..5).map { index -> + async { + controller.execute(key(index), attempt = 1) { + if (entered.incrementAndGet() == 4) allEntered.complete(Unit) + if (index == 2) releaseFast.await() else releaseSlow.await() + } + } + } + withTimeout(1_000) { allEntered.await() } + + val newWorkStarted = CompletableDeferred() + val queued = async { + controller.execute(key(6), attempt = 1) { + newWorkStarted.complete(Unit) + } + } + assertNull(withTimeoutOrNull(50) { newWorkStarted.await() }) + releaseFast.complete(Unit) + withTimeout(1_000) { newWorkStarted.await() } + releaseSlow.complete(Unit) + activeRequests.forEach { it.await() } + queued.await() + } + + assertFalse(controller.snapshot().stopped) + } + + @Test + fun keepsMultipleFailedTaskRetriesSingleFlight() = runBlocking { + val controller = controller(maxConcurrency = 4) + succeed(controller, key(1)) + val failedKeys = listOf(key(2), key(3)) + val allInitialAttemptsEntered = CompletableDeferred() + val releaseInitialAttempts = CompletableDeferred() + val initialEntered = AtomicInteger() + + coroutineScope { + val initialAttempts = (failedKeys + key(4) + key(5)).mapIndexed { index, requestKey -> + async { + runCatching { + controller.execute(requestKey, attempt = 1) { + if (initialEntered.incrementAndGet() == 4) { + allInitialAttemptsEntered.complete(Unit) + } + releaseInitialAttempts.await() + if (index < 2) throw IOException("transient") + } + } + } + } + withTimeout(1_000) { allInitialAttemptsEntered.await() } + releaseInitialAttempts.complete(Unit) + initialAttempts.forEach { it.await() } + + val activeRetries = AtomicInteger() + val peakRetries = AtomicInteger() + val firstRetryEntered = CompletableDeferred() + val releaseRetries = CompletableDeferred() + val retries = failedKeys.map { requestKey -> + async { + controller.execute(requestKey, attempt = 2) { + val current = activeRetries.incrementAndGet() + peakRetries.updateAndGet { previous -> maxOf(previous, current) } + firstRetryEntered.complete(Unit) + releaseRetries.await() + activeRetries.decrementAndGet() + } + } + } + withTimeout(1_000) { firstRetryEntered.await() } + delay(50) + assertEquals(1, peakRetries.get()) + releaseRetries.complete(Unit) + retries.forEach { it.await() } + } + + val stats = controller.snapshot() + assertEquals(1, stats.maxParallelProbes) + assertEquals(0, stats.exhaustedTasks) + assertFalse(stats.stopped) + } + + @Test + fun stopsAfterThreeDifferentTasksExhaustTheirAttempts() = runBlocking { + val controller = controller(maxConcurrency = 128) + + repeat(3) { taskIndex -> + repeat(3) { attemptIndex -> + fail(controller, key(taskIndex + 1), attemptIndex + 1) + } + } + + val stats = controller.snapshot() + assertTrue(stats.stopped) + assertEquals(9, stats.totalAttempts) + assertEquals(9, stats.countedFailures) + assertEquals(8, stats.probeAttempts) + assertEquals(3, stats.exhaustedTasks) + assertFailsWith { + controller.execute(key(4), attempt = 1) { } + } + Unit + } + + @Test + fun onePersistentBadTaskDoesNotStopHealthyWork() = runBlocking { + val controller = controller(maxConcurrency = 128) + repeat(3) { attemptIndex -> fail(controller, key(1), attemptIndex + 1) } + + assertFalse(controller.snapshot().stopped) + succeed(controller, key(2)) + succeed(controller, key(3)) + succeed(controller, key(4)) + + val stats = controller.snapshot() + assertFalse(stats.stopped) + assertEquals(1, stats.exhaustedTasks) + assertEquals(1, stats.recoveredIncidents) + } + + @Test + fun cancellationWhileWaitingDoesNotLeaveAQueuedRequest() = runBlocking { + val controller = controller(maxConcurrency = 1) + val firstEntered = CompletableDeferred() + val releaseFirst = CompletableDeferred() + + coroutineScope { + val first = async { + controller.execute(key(1), attempt = 1) { + firstEntered.complete(Unit) + releaseFirst.await() + } + } + firstEntered.await() + val cancelled = async { + controller.execute(key(2), attempt = 1) { + error("cancelled waiter must not start") + } + } + delay(50) + cancelled.cancelAndJoin() + releaseFirst.complete(Unit) + first.await() + } + + succeed(controller, key(3)) + assertEquals(2, controller.snapshot().totalAttempts) + } + + private fun controller(maxConcurrency: Int) = ProfileDailyRequestController( + maxConcurrentRequests = maxConcurrency, + maxAttempts = 3, + growthFactor = 4, + maxConsecutiveExhaustedTasks = 3, + ) + + private suspend fun succeed( + controller: ProfileDailyRequestController, + key: ProfileDailyRequestKey, + attempt: Int = 1, + ) { + controller.execute(key, attempt) { } + } + + private suspend fun fail( + controller: ProfileDailyRequestController, + key: ProfileDailyRequestKey, + attempt: Int, + ) { + assertFailsWith { + controller.execute(key, attempt) { throw IOException("unavailable") } + } + } + + private fun key(index: Int) = ProfileDailyRequestKey( + groupId = index.toLong(), + startTime = index * 100, + endTime = index * 100 + 50, + ) +} diff --git a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt index 7265d65..ce37c1f 100644 --- a/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt +++ b/src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt @@ -142,6 +142,31 @@ class UserProfileAnalysisServiceTest { assertFalse(UserProfileStore.isConversationProcessed(INPUT_HASH)) } + @Test + fun dailyControllerDoesNotCountSafetyRejectionAsAnOutage() = withProfileStore { + val controller = ProfileDailyRequestController( + maxConcurrentRequests = 128, + maxAttempts = 3, + ) + val model = FakeConversationProfileModel { + throw ModelSafetyRejectionException( + errorType = "invalid_request_error", + errorCode = "cyber_policy", + message = "blocked by policy", + ) + } + + assertFailsWith { + analyze(batch(), model, retryMax = 2, requestController = controller) + } + + val stats = controller.snapshot() + assertEquals(1, stats.totalAttempts) + assertEquals(0, stats.countedFailures) + assertEquals(0, stats.pauseIncidents) + assertFalse(stats.stopped) + } + @Test fun skipsConversationAlreadyProcessedByAnEarlierCall() = withProfileStore { val model = FakeConversationProfileModel { successfulResult() } @@ -250,12 +275,14 @@ class UserProfileAnalysisServiceTest { model: ConversationProfileModel, minAuthoredTextChars: Int = 1, retryMax: Int = 0, + requestController: ProfileDailyRequestController? = null, ): ConversationProfileAnalysisReport? = UserProfileAnalysisService.analyzeConversationBatch( batch = batch, minAuthoredTextChars = minAuthoredTextChars, model = model, retryMax = retryMax, summaryMaxLength = 500, + requestController = requestController, ) private fun successfulResult() = result(