diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 159ac1da..f7c06d79 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -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, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/health/EventStoreLatencyProbe.kt b/apps/server/src/main/kotlin/com/correx/apps/server/health/EventStoreLatencyProbe.kt new file mode 100644 index 00000000..bd57c20c --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/health/EventStoreLatencyProbe.kt @@ -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() } + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt b/apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt new file mode 100644 index 00000000..044b18dd --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt @@ -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() + + 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)", + ) + } + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/health/EventStoreLatencyProbeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/health/EventStoreLatencyProbeTest.kt new file mode 100644 index 00000000..c1ab0fa3 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/health/EventStoreLatencyProbeTest.kt @@ -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) + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt new file mode 100644 index 00000000..ab62f3c4 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt @@ -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, 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) + } +} diff --git a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt index 73e36a9c..6d141a31 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt @@ -23,7 +23,9 @@ data class CorrexConfig( * Continuous health watch (observability-spec §4). When [enabled], a background monitor polls * the on-disk footprint every [intervalMs] and records a degraded/restored event on the edge. * [diskWarnBytes] is the absolute event-log + CAS size that trips a warning; [diskWarnGrowthBytesPerMin] - * is the growth rate that warns log compaction is becoming non-theoretical. Defaults suit a homelab. + * is the growth rate that warns log compaction is becoming non-theoretical. [eventStoreWarnLatencyMs] + * is the event-store latency watermark: a single side-effect-free store read slower than this trips a + * warning that the append/fsync path is contended or the disk is failing. Defaults suit a homelab. */ @Serializable data class HealthConfig( @@ -31,6 +33,7 @@ data class HealthConfig( val intervalMs: Long = 30_000, val diskWarnBytes: Long = 5_000_000_000, val diskWarnGrowthBytesPerMin: Long = 50_000_000, + val eventStoreWarnLatencyMs: Long = 250, ) /**