feat(server): event-store health probe

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.
2026-06-14 18:40:29 +04:00
parent 0957f7c69e
commit 7a1362c973
5 changed files with 214 additions and 1 deletions
@@ -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,
@@ -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
+    }
+}
@@ -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<NewEvent>): List<StoredEvent> = error("unused")
+        override fun read(sessionId: SessionId): List<StoredEvent> = emptyList()
+        override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = emptyList()
+        override fun lastSequence(sessionId: SessionId): Long? = null
+        override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
+        override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
+        override suspend fun lastGlobalSequence(): Long {
+            if (throwOnRead) error("store unavailable")
+            return globalSequence
+        }
+        override fun allEvents(): Sequence<StoredEvent> = emptySequence()
+        override fun allSessionIds(): Set<SessionId> = 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)
+    }
+}
@@ -63,7 +63,8 @@ class LlamaServerHealthProbeTest {
    private class ListEventStore(private val events: MutableList<StoredEvent> = mutableListOf()) : EventStore {
        override suspend fun append(event: NewEvent): StoredEvent = error("unused")
        override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = error("unused")
-        override fun read(sessionId: SessionId): List<StoredEvent> = events.filter { it.metadata.sessionId == sessionId }
+        override fun read(sessionId: SessionId): List<StoredEvent> =
+            events.filter { it.metadata.sessionId == sessionId }
        override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
            events.filter { it.metadata.sessionId == sessionId && it.sequence >= fromSequence }
        override fun lastSequence(sessionId: SessionId): Long? =