From bf36252736ab66f0671afc1d926fd8681450fa13 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 21 Jul 2026 02:01:35 +0400 Subject: [PATCH] feat(health): gate routing event-driven on connection drop, not just periodic poll (#300) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HealthMonitor's periodic poll took ~18s to notice a dead provider — long after a retry had already re-selected it and died. Health needs to GATE routing, not just log reactively. - InferenceRouter.reportFailure(providerId, reason): new default-no-op method; DefaultInferenceRouter implements it by writing Unavailable straight into its health cache, bypassing healthCheck()/TTL entirely. - SessionOrchestrator.runInference: on a connection-level exception (ConnectException/SocketException/IOException or a message matching "prematurely closed"/"connection refused"/"connection reset"), call reportFailure immediately so the very next route() call — the retry driven by #299 — sees the provider as down right away instead of re-selecting it and dying again. Pairs with #299: routing now (1) reacts to a connection drop instantly and (2) waits/backs off for the capability to recover before declaring terminal. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GeyGFXczJb8RUWGBKmkm6G --- .../core/inference/DefaultInferenceRouter.kt | 10 +++++ .../correx/core/inference/InferenceRouter.kt | 8 ++++ .../orchestration/SessionOrchestrator.kt | 20 +++++++++ .../inference/DefaultInferenceRouterTest.kt | 24 ++++++++++ .../SessionOrchestratorIntegrationTest.kt | 45 +++++++++++++++++++ 5 files changed, 107 insertions(+) diff --git a/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt b/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt index b48dd369..c43b114a 100644 --- a/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt +++ b/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt @@ -59,6 +59,16 @@ class DefaultInferenceRouter( fresh } + // Bypasses healthCheck()/TTL entirely — writes Unavailable straight into the cache so the very + // next route() call gates on it, closing the ~18s reactive-poll lag (#300). A later cache-TTL + // expiry or the bounded-wait re-check in route() will naturally pick the provider back up once + // its own healthCheck() reports healthy again. + override suspend fun reportFailure(providerId: ProviderId, reason: String) { + lockFor(providerId).withLock { + mapMutex.withLock { cache[providerId] = HealthEntry(ProviderHealth.Unavailable(reason), timeSource.markNow()) } + } + } + override suspend fun route(stageId: StageId, requiredCapabilities: Set): InferenceProvider { val candidates = requiredCapabilities .flatMap { registry.resolve(it) } diff --git a/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt b/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt index b2436f95..b0924269 100644 --- a/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt +++ b/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt @@ -39,6 +39,14 @@ interface InferenceRouter { requiredCapabilities: Set, modelId: String?, ): InferenceProvider = route(stageId, requiredCapabilities) + + /** + * Event-driven health gate: called the moment a connection-level failure is observed on + * [providerId] (e.g. mid-request connection drop), so the NEXT route() call sees it as + * unavailable immediately instead of waiting for the next periodic health poll/cache TTL to + * catch up. Default no-op for routers that don't cache health. + */ + suspend fun reportFailure(providerId: com.correx.core.events.types.ProviderId, reason: String) = Unit } class NoEligibleProviderException( diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index b98c6ac7..2c01dd7e 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -409,6 +409,13 @@ abstract class SessionOrchestrator( } catch (e: CancellationException) { throw e // never swallow } catch (e: Exception) { + if (isConnectionLevelFailure(e)) { + // Mark it down NOW instead of waiting for the next periodic health poll (~18s lag, + // see #300) — the retry's route() call must see this provider as unavailable + // immediately so it gates/waits (#299) rather than instantly re-selecting the dead + // provider again. + inferenceRouter.reportFailure(provider.id, e.message ?: "connection failure") + } emit( sessionId, InferenceFailedEvent( @@ -423,6 +430,19 @@ abstract class SessionOrchestrator( } } + // Connection-level failures (provider crashed/restarting mid-request) should gate routing + // immediately; other failures (bad request, model error, HTTP 4xx) should not mark the + // provider down since the provider itself is still reachable. + private fun isConnectionLevelFailure(e: Exception): Boolean { + val message = e.message.orEmpty() + return e is java.net.ConnectException || + e is java.net.SocketException || + e is java.io.IOException || // covers ktor/CIO's IOException, which extends java.io.IOException on the JVM + message.contains("prematurely closed", ignoreCase = true) || + message.contains("connection refused", ignoreCase = true) || + message.contains("connection reset", ignoreCase = true) + } + // --- token estimation --- internal open suspend fun estimateTokens(content: String): Int { diff --git a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt index d6284870..244d014c 100644 --- a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt +++ b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt @@ -213,4 +213,28 @@ class DefaultInferenceRouterTest { router.route(stage, setOf(ModelCapability.ToolCalling)) } } + + // ── event-driven health gating (#300) ───────────────────────────────────── + + @Test + fun `reportFailure gates the very next route call without waiting for a health poll`(): Unit = runBlocking { + var healthCheckCount = 0 + val alwaysClaimsHealthy = object : InferenceProvider by provider("a", ModelCapability.General) { + override suspend fun healthCheck(): ProviderHealth { + healthCheckCount++ + return ProviderHealth.Healthy // the provider's own health probe hasn't caught up yet + } + } + val backup = provider("b", ModelCapability.General) + val router = DefaultInferenceRouter(registryOf(alwaysClaimsHealthy, backup), firstStrategy()) + + // Sanity: before reportFailure, routes to the first (still "healthy") provider. + assertSame(alwaysClaimsHealthy, router.route(stage, setOf(ModelCapability.General))) + + router.reportFailure(alwaysClaimsHealthy.id, "connection reset") + + // Immediately after — no delay, no waiting for the next poll — routing must avoid it. + val result = router.route(stage, setOf(ModelCapability.General)) + assertSame(backup, result) + } } diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index cd833972..dec7cb8c 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -43,11 +43,14 @@ import com.correx.core.inference.InferenceProvider import com.correx.core.inference.InferenceRepository import com.correx.core.inference.InferenceRequest import com.correx.core.inference.InferenceResponse +import com.correx.core.inference.DefaultInferenceRouter import com.correx.core.inference.InferenceRouter import com.correx.core.inference.InferenceState import com.correx.core.inference.ModelCapability import com.correx.core.inference.NoEligibleProviderException import com.correx.core.inference.ProviderHealth +import com.correx.core.inference.ProviderRegistry +import com.correx.core.inference.RoutingStrategy import com.correx.core.inference.Token import com.correx.core.inference.TokenUsage import com.correx.core.inference.Tokenizer @@ -253,6 +256,48 @@ class SessionOrchestratorIntegrationTest { assertTrue(events.none { it.payload is WorkflowFailedEvent }) } + @Test + fun `connection-level inference failure gates routing away from the dead provider immediately (#300)`(): Unit = + runBlocking { + val sessionId = SessionId("s2c") + val config = OrchestrationConfig( + retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0), + ) + val graph = threeStageGraph() + + val dead = MockInferenceProvider( + id = ProviderId("dead"), + forcedFailure = "Failed to parse HTTP response: the server prematurely closed the connection", + ) + val backup = MockInferenceProvider(id = ProviderId("backup")) + // Router's own health probe (dead.healthCheck()) still reports Healthy — it hasn't + // polled yet — so only the event-driven reportFailure() from the orchestrator's catch + // block (not the periodic poll) can gate the retry away from `dead`. + val registry = object : ProviderRegistry { + override fun register(provider: com.correx.core.inference.InferenceProvider) = Unit + override fun resolve(capability: ModelCapability) = listOf(dead, backup) + override fun listAll() = listOf(dead, backup) + override suspend fun healthCheckAll() = mapOf(dead.id to ProviderHealth.Healthy, backup.id to ProviderHealth.Healthy) + } + val router = DefaultInferenceRouter( + registry = registry, + strategy = RoutingStrategy { candidates, _ -> candidates.first() }, + ) + + val recoveringOrchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines.copy(inferenceRouter = router), + retryCoordinator = retryCoordinator, + artifactStore = artifactStore, + decisionJournalRepository = decisionJournalRepository, + ) + recoveringOrchestrator.run(sessionId, graph, config) + + val events = eventStore.read(sessionId) + assertNotNull(events.find { it.payload is WorkflowCompletedEvent }) + assertEquals(1, dead.inferCallCount) { "dead provider should only be attempted once, then gated out" } + } + @Test fun `stage that matches no transition retries through the budget before failing`(): Unit = runBlocking { val sessionId = SessionId("s-no-transition-retry")