From 7a1362c9730eb5dcaa62a04e663eb55abc137559 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 14 Jun 2026 18:40:29 +0400 Subject: [PATCH] feat(server): event-store health probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third HealthProbe (after disk + llama): times a cheap lastGlobalSequence() read as an event-store responsiveness heartbeat — DEGRADED on throw or when latency exceeds eventStoreLatencyWarnMs (default 500). Read-only by design (no probe writes into the source-of-truth log; Invariant #1). Plugs into the existing HealthMonitor; no new events. Observability spec §4. Also wraps a long line in LlamaServerHealthProbeTest to clear a detekt MaxLineLength finding introduced in aaf896d. --- .../kotlin/com/correx/apps/server/Main.kt | 4 + .../server/health/EventStoreHealthProbe.kt | 68 +++++++++ .../health/EventStoreHealthProbeTest.kt | 135 ++++++++++++++++++ .../health/LlamaServerHealthProbeTest.kt | 3 +- .../com/correx/core/config/CorrexConfig.kt | 5 + 5 files changed, 214 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/health/EventStoreHealthProbe.kt create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/health/EventStoreHealthProbeTest.kt 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 d23fab0c..9e3647fa 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 @@ -449,6 +449,10 @@ fun main() { livenessTimeoutMs = hc.llamaLivenessTimeoutMs, tpsWarnBelow = hc.llamaTpsWarnBelow, ), + com.correx.apps.server.health.EventStoreHealthProbe( + eventStore = eventStore, + latencyWarnMs = hc.eventStoreLatencyWarnMs, + ), ), intervalMs = hc.intervalMs, initialStatuses = seeded, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/health/EventStoreHealthProbe.kt b/apps/server/src/main/kotlin/com/correx/apps/server/health/EventStoreHealthProbe.kt new file mode 100644 index 00000000..ef1fefc8 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/health/EventStoreHealthProbe.kt @@ -0,0 +1,68 @@ +package com.correx.apps.server.health + +import com.correx.core.events.events.HealthSubject +import com.correx.core.events.stores.EventStore + +/** + * Health probe for the event store (observability-spec §4: EVENT_STORE subject). + * + * **Responsiveness** — every tick times a cheap [EventStore.lastGlobalSequence] read. The + * suspend store call dispatches its own I/O, so the probe times just the read (no extra + * dispatcher hop to inflate the metric). If the call throws (store unavailable) or takes + * longer than [latencyWarnMs], the observation is DEGRADED; otherwise HEALTHY with the + * measured latency as [HealthObservation.observedValue]. + * + * **Why a read, not a write**: The event log is the single source of truth (Invariant #1). + * Writing probe events on every health tick would pollute the journal and grow it + * unbounded. A read gives an equally faithful heartbeat — if the store can't answer a + * sequence query it can't be trusted to record facts. [HealthMonitor] owns edge detection + * and records the observation as a [HealthDegradedEvent] / [HealthRestoredEvent]; this + * probe only observes and returns (Invariants #8/#9). + * + * [nanoTimeSource] is injectable for deterministic testing (fake clocks, no real sleeps). + */ +class EventStoreHealthProbe( + private val eventStore: EventStore, + private val latencyWarnMs: Long = 500, + private val nanoTimeSource: () -> Long = System::nanoTime, +) : HealthProbe { + + override val subject: HealthSubject = HealthSubject.EVENT_STORE + + override suspend fun observe(): HealthObservation { + val startNs = nanoTimeSource() + val result = runCatching { eventStore.lastGlobalSequence() } + val elapsedMs = (nanoTimeSource() - startNs) / NS_PER_MS + + return when { + result.isFailure -> degraded( + metric = "read_latency_ms", + value = elapsedMs, + threshold = latencyWarnMs, + detail = "lastGlobalSequence() threw: ${result.exceptionOrNull()?.message}", + ) + elapsedMs >= latencyWarnMs -> degraded( + metric = "read_latency_ms", + value = elapsedMs, + threshold = latencyWarnMs, + detail = "lastGlobalSequence() took ${elapsedMs}ms (warn ≥ ${latencyWarnMs}ms)", + ) + else -> healthy( + metric = "read_latency_ms", + value = elapsedMs, + threshold = latencyWarnMs, + detail = "event store responsive: lastGlobalSequence() in ${elapsedMs}ms", + ) + } + } + + private fun degraded(metric: String, value: Long, threshold: Long, detail: String) = + HealthObservation(subject, HealthStatus.DEGRADED, metric, value, threshold, detail) + + private fun healthy(metric: String, value: Long, threshold: Long, detail: String) = + HealthObservation(subject, HealthStatus.HEALTHY, metric, value, threshold, detail) + + companion object { + private const val NS_PER_MS = 1_000_000L + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/health/EventStoreHealthProbeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/health/EventStoreHealthProbeTest.kt new file mode 100644 index 00000000..64fb00be --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/health/EventStoreHealthProbeTest.kt @@ -0,0 +1,135 @@ +package com.correx.apps.server.health + +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.SessionId +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.emptyFlow +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 EventStoreHealthProbeTest { + + // --------------------------------------------------------------------------- + // Fake store helpers + // --------------------------------------------------------------------------- + + /** Minimal EventStore that returns a fixed global sequence or throws on demand. */ + private class FakeEventStore( + private val globalSequence: Long = 0L, + private val throwOnRead: Boolean = false, + ) : EventStore { + override suspend fun append(event: NewEvent): StoredEvent = error("unused") + override suspend fun appendAll(events: List): List = error("unused") + override fun read(sessionId: SessionId): List = emptyList() + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = emptyList() + override fun lastSequence(sessionId: SessionId): Long? = null + override fun subscribe(sessionId: SessionId): Flow = emptyFlow() + override fun subscribeAll(): Flow = emptyFlow() + override suspend fun lastGlobalSequence(): Long { + if (throwOnRead) error("store unavailable") + return globalSequence + } + override fun allEvents(): Sequence = emptySequence() + override fun allSessionIds(): Set = emptySet() + } + + /** + * Fake nano-time source that reports [elapsedMs] milliseconds between the first and + * second call — deterministically, with no real sleep. The first call (start) returns 0; + * the second call (after the read) returns [elapsedMs] * NS_PER_MS. + */ + private fun fakeClockMs(elapsedMs: Long): () -> Long { + val ns = elapsedMs * 1_000_000L + var call = 0 + return { if (call++ == 0) 0L else ns } + } + + private fun probe( + store: EventStore = FakeEventStore(), + latencyWarnMs: Long = 500, + nanoTimeSource: () -> Long = fakeClockMs(0L), + ) = EventStoreHealthProbe( + eventStore = store, + latencyWarnMs = latencyWarnMs, + nanoTimeSource = nanoTimeSource, + ) + + // --------------------------------------------------------------------------- + // Tests + // --------------------------------------------------------------------------- + + @Test + fun `healthy when read completes within threshold`(): Unit = runBlocking { + val obs = probe( + store = FakeEventStore(globalSequence = 42L), + latencyWarnMs = 500, + nanoTimeSource = fakeClockMs(elapsedMs = 10), + ).observe() + + assertEquals(HealthStatus.HEALTHY, obs.status) + assertEquals("read_latency_ms", obs.metric) + assertEquals(10L, obs.observedValue) + assertEquals(500L, obs.threshold) + assertTrue(obs.detail.contains("responsive"), "detail should mention 'responsive': ${obs.detail}") + } + + @Test + fun `degraded when read throws`(): Unit = runBlocking { + val obs = probe( + store = FakeEventStore(throwOnRead = true), + latencyWarnMs = 500, + nanoTimeSource = fakeClockMs(elapsedMs = 1), + ).observe() + + assertEquals(HealthStatus.DEGRADED, obs.status) + assertEquals("read_latency_ms", obs.metric) + assertTrue(obs.detail.contains("threw"), "detail should mention 'threw': ${obs.detail}") + assertTrue(obs.detail.contains("store unavailable"), "detail should contain error message: ${obs.detail}") + } + + @Test + fun `degraded when read exceeds latency threshold`(): Unit = runBlocking { + val obs = probe( + store = FakeEventStore(globalSequence = 0L), + latencyWarnMs = 500, + nanoTimeSource = fakeClockMs(elapsedMs = 600), + ).observe() + + assertEquals(HealthStatus.DEGRADED, obs.status) + assertEquals("read_latency_ms", obs.metric) + assertEquals(600L, obs.observedValue) + assertEquals(500L, obs.threshold) + assertTrue(obs.detail.contains("600ms"), "detail should contain elapsed ms: ${obs.detail}") + assertTrue(obs.detail.contains("500ms"), "detail should contain threshold: ${obs.detail}") + } + + @Test + fun `healthy when read latency equals threshold minus one`(): Unit = runBlocking { + // Boundary: exactly one ms under the threshold is still HEALTHY + val obs = probe( + store = FakeEventStore(globalSequence = 1L), + latencyWarnMs = 500, + nanoTimeSource = fakeClockMs(elapsedMs = 499), + ).observe() + + assertEquals(HealthStatus.HEALTHY, obs.status) + assertEquals(499L, obs.observedValue) + } + + @Test + fun `degraded when read latency equals threshold exactly`(): Unit = runBlocking { + // Boundary: exactly at the threshold trips DEGRADED (>= check) + val obs = probe( + store = FakeEventStore(globalSequence = 1L), + latencyWarnMs = 500, + nanoTimeSource = fakeClockMs(elapsedMs = 500), + ).observe() + + assertEquals(HealthStatus.DEGRADED, obs.status) + assertEquals(500L, obs.observedValue) + } +} 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 index 32fdc020..05d10431 100644 --- 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 @@ -63,7 +63,8 @@ class LlamaServerHealthProbeTest { private class ListEventStore(private val events: MutableList = mutableListOf()) : EventStore { override suspend fun append(event: NewEvent): StoredEvent = error("unused") override suspend fun appendAll(events: List): List = error("unused") - override fun read(sessionId: SessionId): List = events.filter { it.metadata.sessionId == sessionId } + override fun read(sessionId: SessionId): List = + events.filter { it.metadata.sessionId == sessionId } override fun readFrom(sessionId: SessionId, fromSequence: Long): List = events.filter { it.metadata.sessionId == sessionId && it.sequence >= fromSequence } override fun lastSequence(sessionId: SessionId): Long? = 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 e7533e9b..30b8206e 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 @@ -29,6 +29,10 @@ data class CorrexConfig( * timeout within this window is reported as DEGRADED (liveness). [llamaTpsWarnBelow] is the * tokens/sec floor below which the probe reports degradation (KV-cache pressure, thermal throttle); * 0.0 disables the degradation dimension (no baseline to compare against on a fresh install). + * + * [eventStoreLatencyWarnMs] is the read-latency threshold above which the event-store probe reports + * DEGRADED. A single cheap read (lastGlobalSequence) is timed each tick; a slow or throwing read + * signals contention or unavailability. 500 ms suits typical SQLite on a homelab SSD. */ @Serializable data class HealthConfig( @@ -38,6 +42,7 @@ data class HealthConfig( val diskWarnGrowthBytesPerMin: Long = 50_000_000, val llamaLivenessTimeoutMs: Long = 5_000, val llamaTpsWarnBelow: Double = 0.0, + val eventStoreLatencyWarnMs: Long = 500, ) /**