feat(health): EventStore latency + llama-server liveness probes

- EventStoreLatencyProbe: times a side-effect-free lastGlobalSequence read, wired into HealthMonitor; HealthConfig.eventStoreWarnLatencyMs (BACKLOG A §4) - LlamaServerHealthProbe: liveness + tokens/sec degradation trend, injectable client (HttpLlamaLivenessClient over Ktor); deterministic unit tests - llama probe left unregistered (TODO) — HttpClient not in monitor scope yet Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 08:39:26 +00:00
parent 784accb08d
commit 266bbf0269
6 changed files with 351 additions and 1 deletions
@@ -436,6 +436,13 @@ fun main() {
                        warnBytes = hc.diskWarnBytes,
                        warnGrowthBytesPerMin = hc.diskWarnGrowthBytesPerMin,
                    ),
+                    com.correx.apps.server.health.EventStoreLatencyProbe.forStore(
+                        eventStore,
+                        hc.eventStoreWarnLatencyMs,
+                    ),
+                    // TODO(wiring): register LlamaServerHealthProbe once an HttpClient + tokens/sec telemetry are
+                    // exposed to the health-monitor scope (currently only built inside
+                    // InfrastructureModule.createModelManager).
                ),
                intervalMs = hc.intervalMs,
                initialStatuses = seeded,
@@ -0,0 +1,65 @@
+package com.correx.apps.server.health
+
+import com.correx.core.events.events.HealthSubject
+import com.correx.core.events.stores.EventStore
+
+private const val NANOS_PER_MILLI = 1_000_000L
+
+/**
+ * Watches event-log append/fsync latency (observability-spec §4: "the organ that must never lie").
+ * Each tick times one durable, side-effect-free store probe and reports DEGRADED when it exceeds
+ * [warnLatencyMs] — a slow store means the append path (fsync) is contended or the disk is failing.
+ *
+ * The timed operation is injected so the probe never itself appends an event (that would both spam
+ * the health log and break replay determinism). [forStore] wires it in production to
+ * [EventStore.lastGlobalSequence], a durable read that rides the same connection an append's fsync
+ * uses. The latency reading is a live environment observation; it never affects replay (Invariant #8).
+ */
+class EventStoreLatencyProbe(
+    private val warnLatencyMs: Long,
+    private val nanoTime: () -> Long = System::nanoTime,
+    private val measure: suspend () -> Unit,
+) : HealthProbe {
+
+    override val subject: HealthSubject = HealthSubject.EVENT_STORE
+
+    override suspend fun observe(): HealthObservation {
+        val start = nanoTime()
+        val failure = runCatching { measure() }.exceptionOrNull()
+        val elapsedMs = (nanoTime() - start) / NANOS_PER_MILLI
+
+        return when {
+            failure != null -> degraded(
+                value = elapsedMs,
+                detail = "event store probe failed: ${failure.message ?: failure::class.simpleName}",
+            )
+            elapsedMs >= warnLatencyMs -> degraded(
+                value = elapsedMs,
+                detail = "event store probe took ${elapsedMs}ms (warn ≥ ${warnLatencyMs}ms)",
+            )
+            else -> HealthObservation(
+                subject = subject,
+                status = HealthStatus.HEALTHY,
+                metric = "latency_ms",
+                observedValue = elapsedMs,
+                threshold = warnLatencyMs,
+                detail = "event store probe took ${elapsedMs}ms",
+            )
+        }
+    }
+
+    private fun degraded(value: Long, detail: String) = HealthObservation(
+        subject = subject,
+        status = HealthStatus.DEGRADED,
+        metric = "latency_ms",
+        observedValue = value,
+        threshold = warnLatencyMs,
+        detail = detail,
+    )
+
+    companion object {
+        /** Production wiring: time a durable, side-effect-free read on [store]. */
+        fun forStore(store: EventStore, warnLatencyMs: Long): EventStoreLatencyProbe =
+            EventStoreLatencyProbe(warnLatencyMs = warnLatencyMs) { store.lastGlobalSequence() }
+    }
+}
@@ -0,0 +1,113 @@
+package com.correx.apps.server.health
+
+import com.correx.core.events.events.HealthSubject
+import io.ktor.client.HttpClient
+import io.ktor.client.request.get
+import io.ktor.http.isSuccess
+
+private const val DEFAULT_DEGRADE_FRACTION = 0.5
+private const val DEFAULT_BASELINE_SAMPLES = 5
+private const val LIVE = 1L
+private const val DEAD = 0L
+
+/** One liveness reading of the llama server: reachable, plus optional throughput telemetry. */
+data class LivenessResult(val reachable: Boolean, val tokensPerSecond: Double? = null)
+
+/** Pings the local inference process; implementations never throw (failures map to unreachable). */
+interface LlamaLivenessClient {
+    suspend fun ping(): LivenessResult
+}
+
+/**
+ * Ktor-backed liveness client: a `GET $baseUrl/health` whose 2xx status is the only signal.
+ * Tokens/sec telemetry is fed separately (the health endpoint reports reachability only), so this
+ * client never populates [LivenessResult.tokensPerSecond]. Any transport exception is swallowed
+ * into an unreachable result so the probe sees a clean signal.
+ */
+class HttpLlamaLivenessClient(
+    private val httpClient: HttpClient,
+    private val baseUrl: String,
+) : LlamaLivenessClient {
+    override suspend fun ping(): LivenessResult = runCatching {
+        val response = httpClient.get("$baseUrl/health")
+        LivenessResult(reachable = response.status.isSuccess())
+    }.getOrElse { LivenessResult(reachable = false) }
+}
+
+/**
+ * Watches the local inference process (observability-spec §4: "liveness + tokens/sec degradation").
+ * Two failure modes: the server stops answering (DEGRADED on the `liveness` metric), or it stays
+ * up but throughput collapses — KV-cache pressure or thermal throttle. The latter is judged against
+ * a rolling PEAK of the last [baselineSamples] readings: when current tps falls below
+ * [degradeFraction] of that peak, the subject degrades on the `tokens_per_sec` metric.
+ *
+ * The rolling window is per-run live state and never affects replay (Invariant #8): the monitor's
+ * recorded edge events are the sole replay input, so a fresh process starting with an empty window
+ * reconstructs identical history from the log.
+ */
+class LlamaServerHealthProbe(
+    private val client: LlamaLivenessClient,
+    private val degradeFraction: Double = DEFAULT_DEGRADE_FRACTION,
+    private val baselineSamples: Int = DEFAULT_BASELINE_SAMPLES,
+) : HealthProbe {
+
+    override val subject: HealthSubject = HealthSubject.LLAMA_SERVER
+
+    private val recent = ArrayDeque<Double>()
+
+    override suspend fun observe(): HealthObservation {
+        val result = runCatching { client.ping() }.getOrElse { LivenessResult(reachable = false) }
+
+        if (!result.reachable) {
+            return HealthObservation(
+                subject = subject,
+                status = HealthStatus.DEGRADED,
+                metric = "liveness",
+                observedValue = DEAD,
+                threshold = LIVE,
+                detail = "llama server unreachable",
+            )
+        }
+
+        val tps = result.tokensPerSecond
+            ?: return HealthObservation(
+                subject = subject,
+                status = HealthStatus.HEALTHY,
+                metric = "liveness",
+                observedValue = LIVE,
+                threshold = LIVE,
+                detail = "llama server reachable",
+            )
+
+        return throughputObservation(tps)
+    }
+
+    private fun throughputObservation(tps: Double): HealthObservation {
+        recent.addLast(tps)
+        while (recent.size > baselineSamples) recent.removeFirst()
+
+        val baseline = recent.maxOrNull() ?: tps
+        val floor = baseline * degradeFraction
+
+        return if (baseline > 0.0 && tps < floor) {
+            HealthObservation(
+                subject = subject,
+                status = HealthStatus.DEGRADED,
+                metric = "tokens_per_sec",
+                observedValue = tps.toLong(),
+                threshold = floor.toLong(),
+                detail = "throughput ${"%.1f".format(tps)} tok/s below ${"%.0f".format(degradeFraction * 100)}% " +
+                    "of peak ${"%.1f".format(baseline)} tok/s",
+            )
+        } else {
+            HealthObservation(
+                subject = subject,
+                status = HealthStatus.HEALTHY,
+                metric = "tokens_per_sec",
+                observedValue = tps.toLong(),
+                threshold = floor.toLong(),
+                detail = "throughput ${"%.1f".format(tps)} tok/s (peak ${"%.1f".format(baseline)} tok/s)",
+            )
+        }
+    }
+}
@@ -0,0 +1,62 @@
+package com.correx.apps.server.health
+
+import com.correx.infrastructure.persistence.InMemoryEventStore
+import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class EventStoreLatencyProbeTest {
+
+    /** Hand-advanced nano source: the [measure] lambda decides how much wall-clock the probe "saw". */
+    private class FakeNanos {
+        var now: Long = 0L
+        fun read(): Long = now
+        fun advanceMs(ms: Long) {
+            now += ms * 1_000_000L
+        }
+    }
+
+    @Test
+    fun `healthy when latency under threshold`(): Unit = runBlocking {
+        val nanos = FakeNanos()
+        val probe = EventStoreLatencyProbe(
+            warnLatencyMs = 50,
+            nanoTime = nanos::read,
+        ) { nanos.advanceMs(10) }
+        val obs = probe.observe()
+        assertEquals(HealthStatus.HEALTHY, obs.status)
+        assertEquals("latency_ms", obs.metric)
+        assertEquals(10, obs.observedValue)
+    }
+
+    @Test
+    fun `degraded when latency crosses threshold`(): Unit = runBlocking {
+        val nanos = FakeNanos()
+        val probe = EventStoreLatencyProbe(
+            warnLatencyMs = 50,
+            nanoTime = nanos::read,
+        ) { nanos.advanceMs(120) }
+        val obs = probe.observe()
+        assertEquals(HealthStatus.DEGRADED, obs.status)
+        assertEquals("latency_ms", obs.metric)
+        assertEquals(120, obs.observedValue)
+        assertEquals(50, obs.threshold)
+    }
+
+    @Test
+    fun `degraded when the timed read throws`(): Unit = runBlocking {
+        val probe = EventStoreLatencyProbe(warnLatencyMs = 50) { error("store down") }
+        val obs = probe.observe()
+        assertEquals(HealthStatus.DEGRADED, obs.status)
+        assertEquals("latency_ms", obs.metric)
+        assertTrue(obs.detail.contains("store down"), "detail should name the failure: ${obs.detail}")
+    }
+
+    @Test
+    fun `forStore reads a real store and stays healthy under a generous watermark`(): Unit = runBlocking {
+        val store = InMemoryEventStore()
+        val probe = EventStoreLatencyProbe.forStore(store, warnLatencyMs = Long.MAX_VALUE)
+        assertEquals(HealthStatus.HEALTHY, probe.observe().status)
+    }
+}
@@ -0,0 +1,100 @@
+package com.correx.apps.server.health
+
+import io.ktor.client.HttpClient
+import io.ktor.client.engine.mock.MockEngine
+import io.ktor.client.engine.mock.respond
+import io.ktor.http.HttpStatusCode
+import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+class LlamaServerHealthProbeTest {
+
+    /** Replays a queued script of liveness readings; [error] is thrown when the queue runs dry on demand. */
+    private class ScriptedClient(results: List<LivenessResult>, private val throwOnPing: Boolean = false) :
+        LlamaLivenessClient {
+        private val queue = ArrayDeque(results)
+        override suspend fun ping(): LivenessResult {
+            if (throwOnPing) error("ping boom")
+            return queue.removeFirst()
+        }
+    }
+
+    private fun reachable(tps: Double? = null) = LivenessResult(reachable = true, tokensPerSecond = tps)
+
+    @Test
+    fun `reachable with stable throughput stays healthy`(): Unit = runBlocking {
+        val probe = LlamaServerHealthProbe(ScriptedClient(listOf(reachable(100.0), reachable(100.0), reachable(100.0))))
+        assertEquals(HealthStatus.HEALTHY, probe.observe().status)
+        assertEquals(HealthStatus.HEALTHY, probe.observe().status)
+        val obs = probe.observe()
+        assertEquals(HealthStatus.HEALTHY, obs.status)
+        assertEquals("tokens_per_sec", obs.metric)
+    }
+
+    @Test
+    fun `unreachable degrades on the liveness metric`(): Unit = runBlocking {
+        val probe = LlamaServerHealthProbe(ScriptedClient(listOf(LivenessResult(reachable = false))))
+        val obs = probe.observe()
+        assertEquals(HealthStatus.DEGRADED, obs.status)
+        assertEquals("liveness", obs.metric)
+        assertEquals(0, obs.observedValue)
+    }
+
+    @Test
+    fun `reachable without telemetry stays healthy on liveness`(): Unit = runBlocking {
+        val probe = LlamaServerHealthProbe(ScriptedClient(listOf(reachable())))
+        val obs = probe.observe()
+        assertEquals(HealthStatus.HEALTHY, obs.status)
+        assertEquals("liveness", obs.metric)
+        assertEquals(1, obs.observedValue)
+    }
+
+    @Test
+    fun `throughput collapse below half of peak degrades, then recovers`(): Unit = runBlocking {
+        val probe = LlamaServerHealthProbe(
+            ScriptedClient(listOf(reachable(100.0), reachable(100.0), reachable(30.0), reachable(100.0))),
+        )
+        assertEquals(HealthStatus.HEALTHY, probe.observe().status)
+        assertEquals(HealthStatus.HEALTHY, probe.observe().status)
+
+        val dropped = probe.observe()
+        assertEquals(HealthStatus.DEGRADED, dropped.status)
+        assertEquals("tokens_per_sec", dropped.metric)
+        assertEquals(30, dropped.observedValue)
+
+        val recovered = probe.observe()
+        assertEquals(HealthStatus.HEALTHY, recovered.status)
+        assertEquals("tokens_per_sec", recovered.metric)
+    }
+
+    @Test
+    fun `ping that throws is treated as unreachable`(): Unit = runBlocking {
+        val probe = LlamaServerHealthProbe(ScriptedClient(emptyList(), throwOnPing = true))
+        val obs = probe.observe()
+        assertEquals(HealthStatus.DEGRADED, obs.status)
+        assertEquals("liveness", obs.metric)
+    }
+
+    private fun clientFor(status: HttpStatusCode) =
+        HttpLlamaLivenessClient(HttpClient(MockEngine { respond(content = "", status = status) }), "http://llama")
+
+    @Test
+    fun `http client reports reachable on 200`(): Unit = runBlocking {
+        assertEquals(true, clientFor(HttpStatusCode.OK).ping().reachable)
+    }
+
+    @Test
+    fun `http client reports unreachable on 503`(): Unit = runBlocking {
+        assertEquals(false, clientFor(HttpStatusCode.ServiceUnavailable).ping().reachable)
+    }
+
+    @Test
+    fun `http client reports unreachable when transport throws`(): Unit = runBlocking {
+        val client = HttpLlamaLivenessClient(
+            HttpClient(MockEngine { error("connection refused") }),
+            "http://llama",
+        )
+        assertEquals(false, client.ping().reachable)
+    }
+}