feat(health): gate routing event-driven on connection drop, not just periodic poll (#300)
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GeyGFXczJb8RUWGBKmkm6G
This commit is contained in:
+24
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user